multi-agent-mux/tests/test_deploy_layout.py

# ==============================================================================
# test_deploy_layout.py — Deploy Script Layout, Refresh & Gitignore Test Suite
# ==============================================================================
import os
import shutil
import subprocess
import tempfile
import unittest

class TestDeployLayout(unittest.TestCase):
    def setUp(self):
        self.repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
        self.temp_dir = tempfile.mkdtemp(prefix="mam_deploy_layout_test_")
        self.test_dir = os.path.join(self.temp_dir, "workspace")
        os.makedirs(self.test_dir, exist_ok=True)
        self.env = os.environ.copy()
        self.env["MAM_REPO_URL"] = self.repo_root
        self.env["MAM_SKIP_VENV"] = "1"
        self.env["MAM_INSTALLER_URL"] = "file://" + os.path.join(self.repo_root, "deploy", "install.sh")

    def tearDown(self):
        shutil.rmtree(self.temp_dir, ignore_errors=True)

    def test_td1_td2_td3_essential_markdowns_only(self):
        """T-D1~D3: Verify essential markdowns are shipped and internal reports/references are excluded."""
        res = subprocess.run(
            ["bash", os.path.join(self.repo_root, "deploy/install.sh"), self.test_dir],
            env=self.env,
            capture_output=True,
            text=True
        )
        self.assertEqual(res.returncode, 0, f"install.sh failed: {res.stderr}")

        # T-D1: No internal reports in .agents/reports
        reports_dir = os.path.join(self.test_dir, ".agents", "reports")
        if os.path.exists(reports_dir):
            reports_files = [f for root, _, files in os.walk(reports_dir) for f in files if f.endswith(".md")]
            self.assertEqual(len(reports_files), 0, f"Internal reports found: {reports_files}")

        # T-D2: Non-essential root docs omitted by default
        self.assertFalse(os.path.exists(os.path.join(self.test_dir, "MESSAGING.md")))
        self.assertFalse(os.path.exists(os.path.join(self.test_dir, "BOOTSTRAP.md")))

        # T-D3: Core essential docs present (B-3 gate)
        self.assertTrue(os.path.exists(os.path.join(self.test_dir, "AGENTS.md")))
        self.assertTrue(os.path.exists(os.path.join(self.test_dir, ".agents", "MULTI_AGENT_RULES.md")))
        self.assertTrue(os.path.exists(os.path.join(self.test_dir, ".agents", "INSTALL.md")))
        self.assertTrue(os.path.exists(os.path.join(self.test_dir, ".agents", "skills", "lib.sh")))

    def test_td6_td7_td8_mam_deploy_layout_and_removal(self):
        """T-D6~D8: Verify remove.sh installed in .mam_deploy (update.sh not, Issue #72) and works properly."""
        subprocess.run(
            ["bash", os.path.join(self.repo_root, "deploy/install.sh"), self.test_dir],
            env=self.env,
            check=True
        )

        # T-D6: Installed in .mam_deploy and executable
        remover = os.path.join(self.test_dir, ".mam_deploy", "remove.sh")
        updater = os.path.join(self.test_dir, ".mam_deploy", "update.sh")
        self.assertTrue(os.path.exists(remover))
        self.assertFalse(os.path.exists(updater))
        self.assertTrue(os.access(remover, os.X_OK))

        # T-D8: Execution from inside .mam_deploy directory targets parent workspace
        res = subprocess.run(
            ["bash", "./remove.sh", "--force"],
            cwd=os.path.join(self.test_dir, ".mam_deploy"),
            capture_output=True,
            text=True
        )
        self.assertEqual(res.returncode, 0, f"remove.sh failed: {res.stderr}")
        self.assertFalse(os.path.exists(os.path.join(self.test_dir, ".agents", "skills")))
        self.assertFalse(os.path.exists(os.path.join(self.test_dir, ".mam_deploy")))

    def test_td12_td13_td14_gitignore_managed_block(self):
        """T-D12~D14: Verify .gitignore managed block injection and manifest exclusion (B-1 gate)."""
        subprocess.run(
            ["bash", os.path.join(self.repo_root, "deploy/install.sh"), self.test_dir],
            env=self.env,
            check=True
        )

        gi_path = os.path.join(self.test_dir, ".gitignore")
        self.assertTrue(os.path.exists(gi_path))

        with open(gi_path) as f:
            content = f.read()

        self.assertIn("# >>> MAM managed block", content)
        self.assertIn("/.mam.env", content)
        self.assertIn("/.mam_deploy/", content)

        # T-D14 [MERGE BLOCKER B-1]: .gitignore MUST NOT be in manifest
        manifest_path = os.path.join(self.test_dir, ".mam", "install_manifest.txt")
        with open(manifest_path) as f:
            manifest = f.read()
        self.assertNotIn(".gitignore", manifest)

    def test_td21_td22_td23_safe_refresh_custom_skills(self):
        """T-D21~D23: Verify custom modifications to skills are preserved on refresh (B-4 gate)."""
        subprocess.run(
            ["bash", os.path.join(self.repo_root, "deploy/install.sh"), self.test_dir],
            env=self.env,
            check=True
        )

        # Modify a framework skill file
        custom_skill = os.path.join(self.test_dir, ".agents", "skills", "lib.sh")
        with open(custom_skill, "a") as f:
            f.write("\n# CUSTOM_USER_MODIFICATION\n")

        # Run install.sh again (REFRESH=1)
        res = subprocess.run(
            ["bash", os.path.join(self.repo_root, "deploy/install.sh"), self.test_dir],
            env=self.env,
            capture_output=True,
            text=True
        )
        self.assertEqual(res.returncode, 0)
        self.assertIn("Local modification detected", res.stderr)

        # Verify modification preserved
        with open(custom_skill) as f:
            self.assertIn("CUSTOM_USER_MODIFICATION", f.read())

        # Verify backup created in .mam/skill-backups
        backups_dir = os.path.join(self.test_dir, ".mam", "skill-backups")
        self.assertTrue(os.path.exists(backups_dir))

    def test_td27_td28_update_preserves_mam_state(self):
        """T-D27~D28: Verify the update skill preserves .mam state files across cycles (B-5 gate)."""
        subprocess.run(
            ["bash", os.path.join(self.repo_root, "deploy/install.sh"), self.test_dir],
            env=self.env,
            check=True
        )

        # Run the update skill (Issue #72: .mam_deploy/update.sh is gone)
        updater = os.path.join(self.test_dir, ".agents", "skills", "multi-agent-mux-update", "scripts", "update.sh")
        env = {k: v for k, v in self.env.items() if not k.startswith("HERDR_")
               and k not in ("WORKSPACE_ROOT", "AGENT_SESSIONS_YAML", "MAM_JOBS_DIR")}
        env["MAM_NO_SUBSCRIBER_DAEMON"] = "1"
        # The update lists the live herdr agents; never ask a real server here.
        fake = os.path.join(self.temp_dir, "fake-herdr-bin")
        os.makedirs(fake)
        with open(os.path.join(fake, "herdr"), "w") as f:
            f.write('#!/bin/sh\necho \'{"sessions": []}\'\n')
        os.chmod(os.path.join(fake, "herdr"), 0o755)
        env["PATH"] = fake + ":" + env["PATH"]
        res = subprocess.run(["bash", updater, "--yes"], cwd=self.test_dir, env=env, capture_output=True, text=True,
                             stdin=subprocess.DEVNULL)
        self.assertEqual(res.returncode, 0, f"the update skill failed: {res.stdout}{res.stderr}")
        self.assertFalse(os.path.exists(os.path.join(self.test_dir, ".mam_deploy", "update.sh")))

        # Verify asset_hashes.txt and version.txt still exist
        self.assertTrue(os.path.exists(os.path.join(self.test_dir, ".mam", "asset_hashes.txt")))
        self.assertTrue(os.path.exists(os.path.join(self.test_dir, ".mam", "version.txt")))

if __name__ == "__main__":
    unittest.main()