libretro/tests/test_artifact_lock.py

#!/usr/bin/env python3
"""Concurrency guard on the pack output directory.

Two runs building the same dist/ leave readers looking at half-written
ZIPs, which surfaces as BadZipFile far from its cause.
"""

from __future__ import annotations

import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "scripts"))

from common import ArtifactLockBusy, artifact_lock, hold_artifact_lock  # noqa: E402


class ArtifactLockTest(unittest.TestCase):
    def setUp(self):
        self.dir = tempfile.mkdtemp()

    def test_writer_excludes_writer(self):
        with artifact_lock(self.dir):
            with self.assertRaises(ArtifactLockBusy):
                with artifact_lock(self.dir):
                    pass

    def test_writer_excludes_reader(self):
        with artifact_lock(self.dir):
            with self.assertRaises(ArtifactLockBusy):
                with artifact_lock(self.dir, exclusive=False):
                    pass

    def test_readers_share(self):
        with artifact_lock(self.dir, exclusive=False):
            with artifact_lock(self.dir, exclusive=False):
                pass

    def test_lock_is_released_on_exit(self):
        with artifact_lock(self.dir):
            pass
        with artifact_lock(self.dir):
            pass

    def test_lock_is_released_on_error(self):
        with self.assertRaises(ValueError):
            with artifact_lock(self.dir):
                raise ValueError("boom")
        with artifact_lock(self.dir):
            pass

    def test_separate_directories_do_not_contend(self):
        other = tempfile.mkdtemp()
        with artifact_lock(self.dir):
            with artifact_lock(other):
                pass

    def test_lock_survives_a_missing_directory(self):
        missing = os.path.join(self.dir, "not-created-yet")
        with artifact_lock(missing):
            self.assertTrue(os.path.isdir(missing))


class PackLockCliTest(unittest.TestCase):
    """generate_pack and pipeline must refuse a directory held elsewhere."""

    def setUp(self):
        self.dir = tempfile.mkdtemp()

    def _run(self, argv: list[str]) -> subprocess.CompletedProcess:
        return subprocess.run(
            [sys.executable, *argv],
            capture_output=True,
            text=True,
            cwd=str(REPO_ROOT),
            timeout=300,
        )

    def test_generate_pack_refuses_locked_output(self):
        with artifact_lock(self.dir):
            proc = self._run(
                [
                    "scripts/generate_pack.py",
                    "--platform",
                    "misterfpga",
                    "--output-dir",
                    self.dir,
                    "--offline",
                ]
            )
        self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
        self.assertIn("is in use by another run", proc.stdout)

    def test_every_writing_mode_refuses_a_locked_output(self):
        """Two of seven modes took the lock; five wrote straight into it.

        A mode that writes packs or manifests into a directory another run
        holds produces the half-written artifact the lock exists to prevent.
        """
        modes = [
            (["--emulator", "handy", "--offline"], "emulator"),
            (["--system", "atari-lynx", "--offline"], "system"),
            (["--manifest-targets"], "manifest-targets"),
            (["--platform", "misterfpga", "--manifest", "--offline"], "manifest"),
            (
                ["--platform", "misterfpga", "--from-md5",
                 "d8f1206299c48946e6ec5ef96d014eaa", "--offline"],
                "from-md5",
            ),
        ]
        with artifact_lock(self.dir):
            for argv, label in modes:
                with self.subTest(mode=label):
                    proc = self._run(
                        ["scripts/generate_pack.py", *argv,
                         "--output-dir", self.dir]
                    )
                    self.assertNotEqual(
                        proc.returncode, 0,
                        f"{label} wrote into a directory held by another run",
                    )
                    self.assertIn("in use", (proc.stdout + proc.stderr).lower())

    def test_verify_packs_refuses_a_writer(self):
        with artifact_lock(self.dir):
            proc = self._run(
                [
                    "scripts/generate_pack.py",
                    "--platform",
                    "misterfpga",
                    "--verify-packs",
                    "--output-dir",
                    self.dir,
                ]
            )
        self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
        self.assertIn("is in use by another run", proc.stdout)

    def test_pipeline_refuses_locked_output(self):
        with artifact_lock(self.dir):
            proc = self._run(
                [
                    "scripts/pipeline.py",
                    "--offline",
                    "--skip-docs",
                    "--output-dir",
                    self.dir,
                ]
            )
        self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr)
        self.assertIn("is in use by another run", proc.stdout)


class HeldLockTest(unittest.TestCase):
    """pipeline.py took the lock step by step: between the pack build and the
    integrity check dist/ was free, another run's purge emptied it, and the
    check reported every platform SKIP and passed."""

    def setUp(self):
        self.dir = tempfile.mkdtemp()

    def _child_locks(self, env: dict) -> subprocess.CompletedProcess:
        code = (
            "import sys; sys.path.insert(0, 'scripts')\n"
            "from common import artifact_lock\n"
            f"with artifact_lock({self.dir!r}):\n    pass\n"
        )
        return subprocess.run(
            [sys.executable, "-c", code], cwd=str(REPO_ROOT), env=env,
            capture_output=True, text=True, timeout=60, check=False,
        )

    def test_a_spawned_step_inherits_the_hold(self):
        with hold_artifact_lock(self.dir):
            self.assertEqual(self._child_locks(dict(os.environ)).returncode, 0)

    def test_any_other_run_is_refused_for_the_whole_hold(self):
        with hold_artifact_lock(self.dir):
            env = {k: v for k, v in os.environ.items() if not k.startswith("RETROBIOS_")}
            proc = self._child_locks(env)
            self.assertNotEqual(proc.returncode, 0)
            self.assertIn("ArtifactLockBusy", proc.stderr)
        self.assertNotIn("RETROBIOS_HELD_ARTIFACT_DIR", os.environ)
        self.assertEqual(self._child_locks(dict(os.environ)).returncode, 0)

    def test_the_pipeline_holds_its_output_for_the_run(self):
        source = (REPO_ROOT / "scripts" / "pipeline.py").read_text(encoding="utf-8")
        self.assertIn("hold_artifact_lock(args.output_dir)", source)


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