tests/test_activate_gap.py

from __future__ import annotations

import importlib.util
import json
import tempfile
import unittest
import zipfile
from pathlib import Path
from unittest import mock

MODULE_PATH = Path(__file__).resolve().parents[1] / "bin" / "activate-gap.py"
SPEC = importlib.util.spec_from_file_location("activate_gap", MODULE_PATH)
MODULE = importlib.util.module_from_spec(SPEC)
assert SPEC and SPEC.loader
SPEC.loader.exec_module(MODULE)


def make_gap(path: Path, version: str, *, valid_manifest: bool = True) -> None:
    with zipfile.ZipFile(path, "w") as archive:
        if valid_manifest:
            payload = {"repository": {"id": "semantic-gap-000", "version": version}}
            archive.writestr("semantic-gap-000/manifest.json", json.dumps(payload))
        else:
            archive.writestr("semantic-gap-000/README.md", "kein Manifest")


class ActivateGapTests(unittest.TestCase):
    def test_existing_path_is_returned_unchanged(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            candidate = root / "semantic-gap.zip"
            candidate.write_bytes(b"test")
            self.assertEqual(candidate.resolve(), MODULE.resolve_candidate(candidate, root / "processed"))

    def test_unique_timestamped_processed_candidate_is_resolved(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            processed = root / "processed"
            processed.mkdir()
            candidate = processed / "20260719-183345-semantic-gap-2.1.45.zip"
            candidate.write_bytes(b"test")
            self.assertEqual(
                candidate.resolve(),
                MODULE.resolve_candidate(root / "semantic-gap-2.1.45.zip", processed),
            )

    def test_missing_processed_candidate_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            with self.assertRaisesRegex(MODULE.GapActivationError, "kein zeitgestempelter Kandidat"):
                MODULE.resolve_candidate(root / "semantic-gap.zip", root / "processed")

    def test_ambiguous_processed_candidates_are_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            processed = root / "processed"
            processed.mkdir()
            for timestamp in ("20260719-183345", "20260719-183346"):
                (processed / f"{timestamp}-semantic-gap.zip").write_bytes(b"test")
            with self.assertRaisesRegex(MODULE.GapActivationError, "mehrdeutig"):
                MODULE.resolve_candidate(root / "semantic-gap.zip", processed)

    def test_first_activation(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            candidate = root / "candidate.zip"
            active = root / "current-gap" / "gap.zip"
            make_gap(candidate, "2.1.5")
            result = MODULE.activate_gap(candidate, active, active.parent / "old")
            self.assertEqual("2.1.5", result["activatedVersion"])
            self.assertTrue(active.exists())
            self.assertIsNone(result["archivedPath"])

    def test_previous_version_is_archived_with_version_and_timestamp(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            candidate = root / "candidate.zip"
            active = root / "current-gap" / "gap.zip"
            active.parent.mkdir(parents=True)
            make_gap(active, "2.1.4")
            make_gap(candidate, "2.1.5")
            with mock.patch.object(MODULE, "_timestamp", return_value="20260718T200000Z"):
                result = MODULE.activate_gap(candidate, active, active.parent / "old")
            archived = active.parent / "old" / "gap-2.1.4-20260718T200000Z.zip"
            self.assertEqual(str(archived), result["archivedPath"])
            self.assertTrue(archived.exists())
            self.assertEqual(("semantic-gap-000", "2.1.5"), MODULE.inspect_gap(active))

    def test_invalid_candidate_preserves_previous_version(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            candidate = root / "candidate.zip"
            active = root / "current-gap" / "gap.zip"
            active.parent.mkdir(parents=True)
            make_gap(active, "2.1.4")
            make_gap(candidate, "2.1.5", valid_manifest=False)
            before = active.read_bytes()
            with self.assertRaises(MODULE.GapActivationError):
                MODULE.activate_gap(candidate, active, active.parent / "old")
            self.assertEqual(before, active.read_bytes())
            self.assertFalse((active.parent / "old").exists())

    def test_path_traversal_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            candidate = Path(tmp) / "candidate.zip"
            with zipfile.ZipFile(candidate, "w") as archive:
                archive.writestr("../manifest.json", json.dumps({"repository": {"id": "x", "version": "1.0"}}))
            with self.assertRaises(MODULE.GapActivationError):
                MODULE.inspect_gap(candidate)


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