98-tests/test_compare_legacy_package.py

import importlib.util
import json
import sys
import tempfile
import unittest
from pathlib import Path

PACKAGING = Path(__file__).parents[1] / "60-packaging"
sys.path.insert(0, str(PACKAGING))
import gap_work_package as GAP

SPEC = importlib.util.spec_from_file_location("compare_legacy_package", PACKAGING / "compare_legacy_package.py")
COMPARE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(COMPARE)


class LegacyPackageComparisonTest(unittest.TestCase):
    def repository(self, name: str, extra=None) -> bytes:
        manifest = {
            "artifactType": name.lower(),
            "artifactName": name,
            "packageClass": "installable",
            "installMode": "overlay",
            "exportMode": "full",
        }
        entries = {
            f"{name}/README.md": f"# {name}\n".encode(),
            f"{name}/manifest.json": GAP._json_bytes(manifest),
        }
        entries.update(extra or {})
        return GAP.build_deterministic_zip(entries)

    def gap(self) -> bytes:
        work_path = "20-status/arbeitspakete/SE-0009.md"
        manifest = {
            "artifactType": "semantic-gap",
            "artifactName": "semantic-gap-000",
            "packageClass": "installable",
            "activeWorkItem": {
                "id": "SE-0009",
                "path": work_path,
                "status": "ready",
                "approved": True,
                "allowedTargetRepositories": ["engineering-tools", "runtime", "workspace"],
            },
            "stateMachine": {"activeWorkItem": "SE-0009"},
        }
        work = (
            "---\nid: SE-0009\nstatus: ready\napproved: true\n"
            'allowed_target_repositories: ["engineering-tools", "runtime", "workspace"]\n'
            "---\n\n# SE-0009\n"
        )
        return GAP.build_deterministic_zip({
            "semantic-gap-000/PROMPT.md": b"# Vertrag\n",
            f"semantic-gap-000/{work_path}": work.encode(),
            "semantic-gap-000/manifest.json": GAP._json_bytes(manifest),
        })

    def filter_config(self, root: Path) -> Path:
        path = root / "filters.json"
        path.write_text(json.dumps({
            "schemaVersion": "1.0",
            "filters": {"runtime": {
                "ruleVersion": "1",
                "applyTo": "temporary-upload-copy-only",
                "exclude": ["runtime/test/**", "runtime/prod/**"],
                "flatRootEquivalent": ["test/**", "prod/**"],
                "reason": "Testzustand filtern.",
                "preserve": ["original-runtime-zip"],
            }},
        }), encoding="utf-8")
        return path

    def fixture(self, root: Path):
        gap = root / "gap.zip"; gap.write_bytes(self.gap())
        sources = {
            "engineering-tools": self.repository("engineering-tools"),
            "runtime": self.repository("runtime", {"runtime/test/result.json": b"test"}),
            "workspace": self.repository("workspace"),
        }
        targets = {name: self.repository(name) for name in sources}
        source_paths = []
        target_paths = []
        for name, data in sources.items():
            path = root / f"source-{name}.zip"; path.write_bytes(data); source_paths.append(f"{name}={path}")
        for name, data in targets.items():
            path = root / f"target-{name}.zip"; path.write_bytes(data); target_paths.append(f"{name}={path}")
        package = root / "new-package"
        GAP.create_work_package(gap, source_paths, target_paths, package, self.filter_config(root))
        legacy = root / "legacy"; legacy.mkdir()
        (legacy / "manifest.json").write_bytes(GAP._json_bytes({
            "artifactType": "chatgpt-upload-set",
            "version": "2.0",
            "exportMode": "source",
            "roundtripId": "TEST",
        }))
        for name, data in sources.items():
            (legacy / f"{name}.zip").write_bytes(data)
        return gap, source_paths, target_paths, package, legacy

    def test_confirms_equivalence_and_manifest_explained_filter(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap, _, targets, package, legacy = self.fixture(root)
            report = COMPARE.compare_packages(legacy, package, gap, targets)
            self.assertEqual(report["status"], "PASSED")
            self.assertEqual(report["unexplainedDifferences"], [])
            runtime = next(item for item in report["repositories"] if item["name"] == "runtime")
            self.assertEqual(runtime["legacyOnlyPaths"], ["runtime/test/result.json"])
            self.assertEqual(runtime["changedPaths"], [])
            self.assertTrue(report["semanticGapPreservedByteIdentically"])

    def test_rejects_unexplained_content_change_without_report(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap, _, targets, package, legacy = self.fixture(root)
            (legacy / "workspace.zip").write_bytes(self.repository("workspace", {"workspace/extra.txt": b"x"}))
            report_path = root / "report.json"
            with self.assertRaisesRegex(GAP.WorkPackageError, "Ungeklärte Inhaltsdifferenz"):
                report = COMPARE.compare_packages(legacy, package, gap, targets)
                COMPARE.write_report(report, report_path)
            self.assertFalse(report_path.exists())

    def test_rejects_modified_target(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap, _, targets, package, legacy = self.fixture(root)
            name, raw_path = targets[0].split("=", 1)
            Path(raw_path).write_bytes(self.repository(name, {f"{name}/changed.txt": b"changed"}))
            with self.assertRaisesRegex(GAP.WorkPackageError, "nicht byteidentisch"):
                COMPARE.compare_packages(legacy, package, gap, targets)

    def test_rejects_repository_set_difference(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap, _, targets, package, legacy = self.fixture(root)
            (legacy / "workspace.zip").unlink()
            with self.assertRaisesRegex(GAP.WorkPackageError, "Repositorymenge unterscheidet"):
                COMPARE.compare_packages(legacy, package, gap, targets)


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