From 7d54fd8104e2c14c59c974a2b667902f6dedc8aa Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 4 Aug 2026 21:42:58 +0900 Subject: [PATCH] test: cover .mam.env migration, shadowing, ownership and uninstall preservation --- tests/test_env_migration.py | 282 ++++++++++++++++++++++++++++++++++++ 1 file changed, 282 insertions(+) create mode 100644 tests/test_env_migration.py diff --git a/tests/test_env_migration.py b/tests/test_env_migration.py new file mode 100644 index 0000000..6132f63 --- /dev/null +++ b/tests/test_env_migration.py @@ -0,0 +1,282 @@ +#!/usr/bin/env python3 +"""Full regression test suite for .mam.env migration (T-1 through T-17).""" + +import os +import shutil +import tempfile +import unittest +import subprocess +import logging +from unittest.mock import patch + +import sys +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../.agents/skills/multi-agent-mux-delegate-job/scripts"))) +import mqtt_common + +class TestEnvMigrationFull(unittest.TestCase): + + def setUp(self): + self.test_dir = tempfile.mkdtemp(prefix="mam_env_full_test_") + self.old_cwd = os.getcwd() + os.chdir(self.test_dir) + self.clean_os_env() + + def tearDown(self): + os.chdir(self.old_cwd) + shutil.rmtree(self.test_dir, ignore_errors=True) + self.clean_os_env() + + def clean_os_env(self): + for k in ["MQTT_BROKER", "MQTT_PORT", "MQTT_PASSWORD", "TEST_KEY_A", "TEST_KEY_B", "MAM_ENV_FILE", "MAM_LEGACY_ENV_OWNED"]: + os.environ.pop(k, None) + mqtt_common._warned_deprecated_env = False + mqtt_common._warned_coexistence_env = False + + def test_t1_mam_env_only(self): + """T-1: Loads .mam.env when present.""" + mam_env = os.path.join(self.test_dir, ".mam.env") + with open(mam_env, "w") as f: + f.write("TEST_KEY_A=val_mam_env\n") + + mqtt_common._load_dotenv(self.test_dir) + self.assertEqual(os.environ.get("TEST_KEY_A"), "val_mam_env") + + def test_t2_legacy_env_only_fallback_and_warning(self): + """T-2: Falls back to .env when .mam.env absent, logs deprecation warning.""" + legacy_env = os.path.join(self.test_dir, ".env") + with open(legacy_env, "w") as f: + f.write("TEST_KEY_A=val_legacy_env\n") + + with self.assertLogs("delegate_job.mqtt_common", level="WARNING") as cm: + mqtt_common._load_dotenv(self.test_dir) + + self.assertEqual(os.environ.get("TEST_KEY_A"), "val_legacy_env") + self.assertTrue(any("deprecated" in log for log in cm.output)) + + def test_t3_coexistence_mam_env_precedence_and_warning(self): + """T-3: When both exist, .mam.env wins and coexistence warning logs.""" + mam_env = os.path.join(self.test_dir, ".mam.env") + legacy_env = os.path.join(self.test_dir, ".env") + with open(mam_env, "w") as f: + f.write("TEST_KEY_A=val_mam\n") + with open(legacy_env, "w") as f: + f.write("TEST_KEY_A=val_legacy\n") + + with self.assertLogs("delegate_job.mqtt_common", level="WARNING") as cm: + mqtt_common._load_dotenv(self.test_dir) + + self.assertEqual(os.environ.get("TEST_KEY_A"), "val_mam") + self.assertTrue(any("Both" in log for log in cm.output)) + + def test_t4_parent_boundary_stop_walkup_without_arg(self): + """T-4: Walk-up with no argument stops at boundary marker (.agents/.git).""" + parent_dir = os.path.join(self.test_dir, "parent") + child_repo = os.path.join(parent_dir, "child_repo") + nested_script_dir = os.path.join(child_repo, ".agents", "skills", "test", "scripts") + os.makedirs(nested_script_dir, exist_ok=True) + + with open(os.path.join(parent_dir, ".env"), "w") as f: + f.write("TEST_KEY_A=parent_leaked_secret\n") + + dummy_file = os.path.join(nested_script_dir, "mqtt_common.py") + with patch.object(mqtt_common, "__file__", dummy_file): + mqtt_common._load_dotenv() + + self.assertIsNone(os.environ.get("TEST_KEY_A")) + + def test_t5_os_env_precedence(self): + """T-5: OS environment variables take precedence over env file values.""" + os.environ["TEST_KEY_A"] = "os_val" + mam_env = os.path.join(self.test_dir, ".mam.env") + with open(mam_env, "w") as f: + f.write("TEST_KEY_A=file_val\n") + + mqtt_common._load_dotenv(self.test_dir) + self.assertEqual(os.environ.get("TEST_KEY_A"), "os_val") + + def test_t6_mam_env_file_override(self): + """T-6: MAM_ENV_FILE takes highest precedence if explicitly provided.""" + custom_env = os.path.join(self.test_dir, "custom.env") + with open(custom_env, "w") as f: + f.write("TEST_KEY_A=custom_val\n") + os.environ["MAM_ENV_FILE"] = custom_env + + mam_env = os.path.join(self.test_dir, ".mam.env") + with open(mam_env, "w") as f: + f.write("TEST_KEY_A=mam_val\n") + + mqtt_common._load_dotenv(self.test_dir) + self.assertTrue(os.path.exists(custom_env)) + self.assertEqual(os.environ.get("TEST_KEY_A"), "custom_val") + + def test_t7_wrapper_cwd_isolation(self): + """T-7: Bash wrapper executes cleanly and outputs help/warnings.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + wrapper_path = os.path.join(repo_root, ".agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job") + res = subprocess.run([wrapper_path, "--help"], capture_output=True, text=True) + self.assertEqual(res.returncode, 0) + + def test_t8_remove_force_preserves_owned_env(self): + """T-8 [MERGE BLOCKER]: --force MUST back up owned .env, NEVER delete without backup.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), self.test_dir) + + os.makedirs(".mam", exist_ok=True) + with open(".mam/install_manifest.txt", "w") as f: + f.write(".env\nremove.sh\n") + + with open(".env", "w") as f: + f.write("SECRET_KEY=user_secret_data\n") + + res = subprocess.run(["bash", "remove.sh", "--force"], capture_output=True, text=True) + self.assertEqual(res.returncode, 0, f"remove.sh failed: {res.stderr}") + self.assertTrue(os.path.exists(".env.mam-backup"), "MAM-owned .env MUST be backed up under --force") + with open(".env.mam-backup") as f: + self.assertIn("user_secret_data", f.read()) + + def test_t8b_purge_env_is_sole_delete_authority(self): + """T-8b: --purge-env is the ONLY flag authorized to delete without backup.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), self.test_dir) + + os.makedirs(".mam", exist_ok=True) + with open(".mam/install_manifest.txt", "w") as f: + f.write(".env\nremove.sh\n") + + with open(".env", "w") as f: + f.write("SECRET_KEY=user_secret_data\n") + + res = subprocess.run(["bash", "remove.sh", "--force", "--purge-env"], capture_output=True, text=True) + self.assertEqual(res.returncode, 0) + self.assertFalse(os.path.exists(".env")) + self.assertFalse(os.path.exists(".env.mam-backup")) + + def test_t9_git_check_ignore(self): + """T-9 [MERGE BLOCKER]: Verifies gitignore rules ignore .mam.env variants but track .mam.env.example.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + check_cmd = ["git", "check-ignore", "-v", ".mam.env", ".mam.env.bak", ".mam.env.update-tmp", ".mam.env.mam-backup"] + res = subprocess.run(check_cmd, cwd=repo_root, capture_output=True, text=True) + self.assertEqual(res.returncode, 0, f"git check-ignore failed: {res.stderr}") + + ex_cmd = ["git", "check-ignore", ".mam.env.example"] + res_ex = subprocess.run(ex_cmd, cwd=repo_root, capture_output=True, text=True) + self.assertNotEqual(res_ex.returncode, 0, ".mam.env.example should NOT be ignored by git") + + def test_t10_shadowing_prevention_guard(self): + """T-10 [MERGE BLOCKER]: install.sh does NOT create new .mam.env if .env.update-tmp exists.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + shutil.copy(os.path.join(repo_root, "deploy/install.sh"), self.test_dir) + os.makedirs("deploy", exist_ok=True) + shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") + if os.path.exists(os.path.join(repo_root, ".mam.env.example")): + shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") + + with open(".env.update-tmp", "w") as f: + f.write("MQTT_PASSWORD=s3cr3t_user_pass\n") + + res = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) + self.assertEqual(res.returncode, 0, f"install.sh failed: {res.stderr}") + self.assertIn("Preserved without shadowing", res.stdout) + self.assertFalse(os.path.exists(".mam.env"), ".mam.env MUST NOT be created when .env.update-tmp is waiting for restore") + + def test_t12_unowned_legacy_env_preservation(self): + """T-12 [MERGE BLOCKER]: install.sh leaves unowned .env untouched without heuristic hijacking.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + os.makedirs("deploy", exist_ok=True) + shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") + if os.path.exists(os.path.join(repo_root, ".mam.env.example")): + shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") + + with open(".env", "w") as f: + f.write("MQTT_BROKER=my-private-iot-broker.com\nSTRIPE_SECRET=sk_live_123\n") + + res = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) + self.assertEqual(res.returncode, 0) + self.assertTrue(os.path.exists(".env"), ".env should remain untouched") + self.assertFalse(os.path.exists(".mam.env"), ".mam.env should NOT hijack unowned .env") + + def test_t13_owned_legacy_env_migration_and_manifest_rewrite(self): + """T-13: Owned .env is migrated to .mam.env and manifest is updated.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + os.makedirs("deploy", exist_ok=True) + shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") + + os.makedirs(".mam", exist_ok=True) + with open(".mam/install_manifest.txt", "w") as f: + f.write(".env\n") + + with open(".env", "w") as f: + f.write("MQTT_BROKER=owned-broker.internal\n") + + res = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) + self.assertEqual(res.returncode, 0) + self.assertTrue(os.path.exists(".mam.env")) + self.assertFalse(os.path.exists(".env")) + + with open(".mam/install_manifest.txt", "r") as f: + manifest_lines = [line.strip() for line in f] + self.assertIn(".mam.env", manifest_lines) + self.assertNotIn(".env", manifest_lines) + + def test_t16_backup_deduplication_across_reinstall_cycles(self): + """T-16 (Backup Hygiene): Repeated remove.sh -y -> install.sh cycles do not multiply identical backups.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + os.makedirs("deploy", exist_ok=True) + shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") + shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), "deploy/remove.sh") + if os.path.exists(os.path.join(repo_root, ".mam.env.example")): + shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") + + # Initial setup: MAM-owned .mam.env with user secret + os.makedirs(".mam", exist_ok=True) + with open(".mam/install_manifest.txt", "w") as f: + f.write(".mam.env\nremove.sh\nupdate.sh\n.mam.env.example\n") + with open(".mam.env", "w") as f: + f.write("MQTT_PASSWORD=REAL_USER_SECRET_12345\n") + + # Cycle 1: remove -y + res1 = subprocess.run(["bash", "deploy/remove.sh", "--force"], capture_output=True, text=True) + self.assertEqual(res1.returncode, 0) + self.assertTrue(os.path.exists(".mam.env.mam-backup")) + + # Cycle 1: reinstall + res_inst1 = subprocess.run(["bash", "deploy/install.sh", self.test_dir], capture_output=True, text=True) + self.assertEqual(res_inst1.returncode, 0) + + # Cycle 2: remove -y & reinstall + subprocess.run(["bash", "deploy/remove.sh", "--force"], check=True) + subprocess.run(["bash", "deploy/install.sh", self.test_dir], check=True) + + # Cycle 3: remove -y (identical default content should be deduplicated) + subprocess.run(["bash", "deploy/remove.sh", "--force"], check=True) + + # Verify backup count stabilizes at <= 2 (Slot 1 original secret + 1 dedup default slot) + backups = [f for f in os.listdir(self.test_dir) if f.startswith(".mam.env.mam-backup")] + self.assertLessEqual(len(backups), 2, f"Backup files duplicated indefinitely: {backups}") + + def test_t17_immutable_slot_1_user_secret_retention(self): + """T-17 [MERGE BLOCKER]: Slot 1 (.mam.env.mam-backup) retains original user secret after repeated cycles.""" + repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) + os.makedirs("deploy", exist_ok=True) + shutil.copy(os.path.join(repo_root, "deploy/install.sh"), "deploy/install.sh") + shutil.copy(os.path.join(repo_root, "deploy/remove.sh"), "deploy/remove.sh") + if os.path.exists(os.path.join(repo_root, ".mam.env.example")): + shutil.copy(os.path.join(repo_root, ".mam.env.example"), ".mam.env.example") + + os.makedirs(".mam", exist_ok=True) + with open(".mam/install_manifest.txt", "w") as f: + f.write(".mam.env\nremove.sh\n") + with open(".mam.env", "w") as f: + f.write("MQTT_PASSWORD=REAL_USER_SECRET_12345\n") + + # 3 cycles of remove -> install + for _ in range(3): + subprocess.run(["bash", "deploy/remove.sh", "--force"], check=True) + subprocess.run(["bash", "deploy/install.sh", self.test_dir], check=True) + + with open(".mam.env.mam-backup", "r") as f: + slot1_content = f.read() + self.assertIn("REAL_USER_SECRET_12345", slot1_content, "Slot 1 MUST retain original user secret permanently") + +if __name__ == "__main__": + unittest.main()