283 lines
14 KiB
Python
283 lines
14 KiB
Python
#!/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()
|