98-tests/test_gap_work_package.py

import hashlib
import importlib.util
import json
import tempfile
import unittest
import zipfile
from io import BytesIO
from pathlib import Path

PACKAGING = Path(__file__).parents[1] / "60-packaging"
SPEC = importlib.util.spec_from_file_location("gap_work_package", PACKAGING / "gap_work_package.py")
MODULE = importlib.util.module_from_spec(SPEC)
import sys
sys.path.insert(0, str(PACKAGING))
SPEC.loader.exec_module(MODULE)


class GapWorkPackageTest(unittest.TestCase):
    def repository(self, name: str, root_name: str | None = None, extra: dict[str, bytes] | None = None, flat: bool = False) -> bytes:
        root = root_name or name
        manifest = {
            "artifactType": name.lower(),
            "artifactName": name,
            "packageClass": "installable",
            "installMode": "overlay",
            "exportMode": "full",
        }
        prefix = "" if flat else f"{root}/"
        entries = {
            f"{prefix}README.md": f"# {name}\n".encode(),
            f"{prefix}manifest.json": MODULE._json_bytes(manifest),
        }
        entries.update(extra or {})
        return MODULE.build_deterministic_zip(entries)

    def filter_config(self, root: Path) -> Path:
        path = root / "upload-filters.json"
        path.write_text(json.dumps({
            "schemaVersion": "1.0",
            "filters": {
                "runtime": {
                    "ruleVersion": "2",
                    "applyTo": "temporary-upload-copy-only",
                    "exclude": ["runtime/test/**", "runtime/prod/**", "runtime/production/**"],
                    "flatRootEquivalent": ["test/**", "prod/**", "production/**"],
                    "reason": "Nur die temporaere Uploadkopie filtern.",
                    "preserve": [
                        "local-runtime-repository",
                        "original-runtime-zip",
                        "returned-runtime-roundtrip-zip",
                    ],
                }
            },
        }), encoding="utf-8")
        return path

    def gap(self, allowed=None) -> bytes:
        allowed = allowed or ["engineering-tools", "runtime"]
        path = "20-status/arbeitspakete/SE-0006.md"
        manifest = {
            "artifactType": "semantic-gap",
            "artifactName": "semantic-gap-000",
            "packageClass": "installable",
            "activeWorkItem": {
                "id": "SE-0006",
                "path": path,
                "status": "ready",
                "approved": True,
                "allowedTargetRepositories": allowed,
            },
            "stateMachine": {"activeWorkItem": "SE-0006"},
        }
        work_item = (
            "---\n"
            "id: SE-0006\n"
            "status: ready\n"
            "approved: true\n"
            f"allowed_target_repositories: {json.dumps(allowed)}\n"
            "---\n\n# SE-0006\n\nImplementieren.\n"
        )
        return MODULE.build_deterministic_zip({
            "semantic-gap-000/PROMPT.md": b"# Grundvertrag\n\nEin Sprint.\n",
            f"semantic-gap-000/{path}": work_item.encode(),
            "semantic-gap-000/manifest.json": MODULE._json_bytes(manifest),
        })

    def create_fixture(self, root: Path, target_name="engineering-tools"):
        gap = root / "gap.zip"
        source = root / "source.zip"
        target = root / "target.zip"
        gap.write_bytes(self.gap())
        source.write_bytes(self.repository("SolutionArchitecture"))
        target.write_bytes(self.repository(target_name))
        return gap, source, target

    def test_deterministic_structure_prompt_manifest_and_immutability(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap, source, target = self.create_fixture(root)
            original_hashes = [hashlib.sha256(path.read_bytes()).hexdigest() for path in (gap, source, target)]
            first, second = root / "first", root / "second"
            source_spec = [f"SolutionArchitecture={source}"]
            target_spec = [f"engineering-tools={target}"]
            MODULE.create_work_package(gap, source_spec, target_spec, first)
            MODULE.create_work_package(gap, source_spec, target_spec, second)
            for name in ("package-manifest.json", "SHA256SUMS.txt", "sources.zip", "targets.zip"):
                self.assertEqual((first / name).read_bytes(), (second / name).read_bytes())
            self.assertEqual(original_hashes, [hashlib.sha256(path.read_bytes()).hexdigest() for path in (gap, source, target)])
            manifest = MODULE.validate_work_package(first)
            self.assertEqual(manifest["activeWorkItem"]["id"], "SE-0006")
            self.assertEqual(sorted(path.name for path in first.iterdir()), ["SHA256SUMS.txt", "package-manifest.json", "sources.zip", "targets.zip"])
            self.assertEqual(manifest["repositoryRoles"]["sources"][0]["role"], "immutable-source")
            self.assertEqual(manifest["repositoryRoles"]["targets"][0]["role"], "mutable-target")
            self.assertEqual(manifest["repositoryRoles"]["targets"][0]["identity"]["artifactName"], "engineering-tools")
            checksums = (first / "SHA256SUMS.txt").read_text(encoding="utf-8").splitlines()
            self.assertEqual(len(checksums), 7)
            self.assertTrue(any(line.endswith("  sources.zip/SolutionArchitecture.zip") for line in checksums))
            self.assertTrue(any(line.endswith("  targets.zip/gap.zip") for line in checksums))
            with zipfile.ZipFile(first / "sources.zip") as sources:
                self.assertEqual(sources.namelist(), ["PROMPT.md", "SolutionArchitecture.zip"])
                prompt = sources.read("PROMPT.md").decode()
                self.assertIn("# Grundvertrag", prompt)
                self.assertIn("# Aktiviertes Work Item", prompt)
                self.assertIn("# SE-0006", prompt)
                self.assertEqual(sources.read("SolutionArchitecture.zip"), source.read_bytes())
            with zipfile.ZipFile(first / "targets.zip") as targets:
                self.assertEqual(targets.namelist(), ["engineering-tools.zip", "gap.zip"])
                self.assertEqual(targets.read("gap.zip"), gap.read_bytes())

    def test_rejects_target_outside_allowed_scope(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap(["engineering-tools"]))
            target = root / "workspace.zip"; target.write_bytes(self.repository("workspace"))
            with self.assertRaisesRegex(MODULE.WorkPackageError, "Allowed Target Scope"):
                MODULE.create_work_package(gap, [], [f"workspace={target}"], root / "result")

    def test_accepts_allowed_scope_with_different_order(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap_bytes = self.gap(["workspace", "runtime", "engineering-tools"])
            with zipfile.ZipFile(BytesIO(gap_bytes)) as archive:
                entries = {name: archive.read(name) for name in archive.namelist()}
            work_path = "semantic-gap-000/20-status/arbeitspakete/SE-0006.md"
            entries[work_path] = entries[work_path].replace(
                b'["workspace", "runtime", "engineering-tools"]',
                b'["engineering-tools", "workspace", "runtime"]',
            )
            gap = root / "gap.zip"
            gap.write_bytes(MODULE.build_deterministic_zip(entries))
            target = root / "workspace.zip"
            target.write_bytes(self.repository("workspace"))
            MODULE.create_work_package(gap, [], [f"workspace={target}"], root / "result")

    def test_rejects_repository_identity_mismatch(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            target = root / "target.zip"; target.write_bytes(self.repository("runtime"))
            with self.assertRaisesRegex(MODULE.WorkPackageError, "Repository-Identität"):
                MODULE.create_work_package(gap, [], [f"engineering-tools={target}"], root / "result")

    def test_rejects_unsafe_zip_member(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            target = root / "target.zip"
            with zipfile.ZipFile(target, "w") as archive:
                archive.writestr("../manifest.json", "{}")
            with self.assertRaisesRegex(MODULE.WorkPackageError, "unsicherer ZIP-Pfad"):
                MODULE.create_work_package(gap, [], [f"engineering-tools={target}"], root / "result")
            self.assertFalse((root / "result").exists())

    def test_rejects_duplicate_zip_member(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            target = root / "target.zip"
            with zipfile.ZipFile(target, "w") as archive:
                archive.writestr("engineering-tools/manifest.json", MODULE._json_bytes({"artifactType": "engineering-tools"}))
                archive.writestr("engineering-tools/manifest.json", MODULE._json_bytes({"artifactType": "engineering-tools"}))
            with self.assertRaisesRegex(MODULE.WorkPackageError, "doppelter ZIP-Pfad"):
                MODULE.create_work_package(gap, [], [f"engineering-tools={target}"], root / "result")
            self.assertFalse((root / "result").exists())

    def test_rejects_unexpected_repository_root(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            target = root / "target.zip"; target.write_bytes(self.repository("engineering-tools", root_name="unexpected"))
            with self.assertRaisesRegex(MODULE.WorkPackageError, "unerwarteter Repository-Root"):
                MODULE.create_work_package(gap, [], [f"engineering-tools={target}"], root / "result")

    def test_rejects_corrupted_archive(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            data = bytearray(self.repository("engineering-tools"))
            with zipfile.ZipFile(BytesIO(data)) as archive:
                info = archive.getinfo("engineering-tools/README.md")
                offset = info.header_offset + 30 + len(info.filename.encode()) + len(info.extra)
            data[offset + 1] ^= 0xFF
            target = root / "target.zip"; target.write_bytes(data)
            with self.assertRaisesRegex(MODULE.WorkPackageError, "beschädigtes ZIP"):
                MODULE.create_work_package(gap, [], [f"engineering-tools={target}"], root / "result")
            self.assertFalse((root / "result").exists())

    def test_validation_rejects_checksum_tampering(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap, source, target = self.create_fixture(root)
            output = root / "result"
            MODULE.create_work_package(
                gap, [f"SolutionArchitecture={source}"], [f"engineering-tools={target}"], output
            )
            checksums = (output / "SHA256SUMS.txt").read_text(encoding="utf-8")
            (output / "SHA256SUMS.txt").write_text("0" + checksums[1:], encoding="utf-8")
            with self.assertRaisesRegex(MODULE.WorkPackageError, "SHA256SUMS"):
                MODULE.validate_work_package(output)

    def test_existing_output_is_preserved(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap, source, target = self.create_fixture(root)
            output = root / "result"; output.mkdir()
            sentinel = output / "sentinel.txt"; sentinel.write_text("keep", encoding="utf-8")
            with self.assertRaisesRegex(MODULE.WorkPackageError, "existiert bereits"):
                MODULE.create_work_package(
                    gap, [f"SolutionArchitecture={source}"], [f"engineering-tools={target}"], output
                )
            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")

    def test_filters_wrapped_runtime_source_and_preserves_original_and_target(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            runtime = self.repository("runtime", extra={
                "runtime/test/report.json": b"test",
                "runtime/prod/state.json": b"prod",
                "runtime/production/README.md": b"production",
                "runtime/config/app.json": b"config",
            })
            source = root / "runtime-source.zip"; source.write_bytes(runtime)
            target = root / "runtime-target.zip"; target.write_bytes(runtime)
            before = hashlib.sha256(source.read_bytes()).hexdigest()
            output = root / "result"
            MODULE.create_work_package(
                gap,
                [f"runtime={source}"],
                [f"runtime={target}"],
                output,
                self.filter_config(root),
            )
            self.assertEqual(before, hashlib.sha256(source.read_bytes()).hexdigest())
            with zipfile.ZipFile(output / "sources.zip") as outer_sources:
                filtered_runtime = outer_sources.read("runtime.zip")
            with zipfile.ZipFile(BytesIO(filtered_runtime)) as filtered:
                names = filtered.namelist()
                self.assertNotIn("runtime/test/report.json", names)
                self.assertNotIn("runtime/prod/state.json", names)
                self.assertNotIn("runtime/production/README.md", names)
                self.assertIn("runtime/config/app.json", names)
            with zipfile.ZipFile(output / "targets.zip") as outer_targets:
                with zipfile.ZipFile(BytesIO(outer_targets.read("runtime.zip"))) as filtered_target:
                    self.assertNotIn("runtime/test/report.json", filtered_target.namelist())
                    self.assertNotIn("runtime/prod/state.json", filtered_target.namelist())
                    self.assertNotIn("runtime/production/README.md", filtered_target.namelist())
                    self.assertIn("runtime/config/app.json", filtered_target.namelist())
            manifest = json.loads((output / "package-manifest.json").read_text(encoding="utf-8"))
            evidence = manifest["uploadFilters"][0]
            self.assertEqual(evidence["ruleVersion"], "2")
            self.assertEqual(evidence["rootForm"], "wrapped")
            self.assertEqual(evidence["reason"], "Nur die temporaere Uploadkopie filtern.")
            self.assertEqual(evidence["excludedPaths"], [
                "runtime/prod/state.json",
                "runtime/production/README.md",
                "runtime/test/report.json",
            ])

    def test_default_runtime_filter_excludes_recursive_root_and_generated_outputs(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            runtime = root / "runtime.zip"
            runtime.write_bytes(self.repository("runtime", extra={
                "runtime/runtime/manifest.json": MODULE._json_bytes({"artifactType": "runtime", "artifactName": "runtime"}),
                "runtime/runtime/bin/publish-reports-entrypoint.py": b"nested copy",
                "runtime/reports/index.html": b"generated report",
                "runtime/reports/SE-0057-semantic-publisher-input.json": b'{"semanticLanding": {}}',
                "runtime/web/reports/index.html": b"published report",
                "runtime/cache/state.json": b"cache",
                "runtime/bin/publish-reports-entrypoint.py": b"source entrypoint",
            }))
            target = root / "target.zip"; target.write_bytes(self.repository("runtime"))
            output = root / "result"
            config = PACKAGING / "upload-filters.json"
            MODULE.create_work_package(
                gap, [f"runtime={runtime}"], [f"runtime={target}"], output, config
            )
            with zipfile.ZipFile(output / "sources.zip") as outer:
                with zipfile.ZipFile(BytesIO(outer.read("runtime.zip"))) as filtered:
                    names = filtered.namelist()
            self.assertNotIn("runtime/runtime/manifest.json", names)
            self.assertNotIn("runtime/runtime/bin/publish-reports-entrypoint.py", names)
            self.assertNotIn("runtime/reports/index.html", names)
            self.assertIn("runtime/reports/SE-0057-semantic-publisher-input.json", names)
            self.assertNotIn("runtime/web/reports/index.html", names)
            self.assertNotIn("runtime/cache/state.json", names)
            self.assertIn("runtime/bin/publish-reports-entrypoint.py", names)

    def test_default_engineering_tools_filter_excludes_recursive_root(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            tools = root / "engineering-tools.zip"
            tools.write_bytes(self.repository("engineering-tools", extra={
                "engineering-tools/engineering-tools/manifest.json": MODULE._json_bytes({"artifactType": "engineering-tools", "artifactName": "engineering-tools"}),
                "engineering-tools/engineering-tools/60-packaging/gap_work_package.py": b"nested copy",
                "engineering-tools/60-packaging/gap_work_package.py": b"source packager",
            }))
            target = root / "target.zip"; target.write_bytes(self.repository("engineering-tools"))
            output = root / "result"
            MODULE.create_work_package(
                gap, [f"engineering-tools={tools}"], [f"engineering-tools={target}"], output,
                PACKAGING / "upload-filters.json",
            )
            with zipfile.ZipFile(output / "sources.zip") as outer:
                with zipfile.ZipFile(BytesIO(outer.read("engineering-tools.zip"))) as filtered:
                    names = filtered.namelist()
            self.assertNotIn("engineering-tools/engineering-tools/manifest.json", names)
            self.assertNotIn("engineering-tools/engineering-tools/60-packaging/gap_work_package.py", names)
            self.assertIn("engineering-tools/60-packaging/gap_work_package.py", names)

    def test_default_workspace_filter_excludes_archive_from_source_and_target(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap(["workspace"]))
            workspace = root / "workspace.zip"
            workspace.write_bytes(self.repository("workspace", extra={
                "workspace/archive/roundtrip-imports/previous.zip": b"local archive",
                "workspace/deployment/local/bin/create-chatgpt-upload-zips.sh": b"source entrypoint",
            }))
            output = root / "result"
            MODULE.create_work_package(
                gap, [f"workspace={workspace}"], [f"workspace={workspace}"], output,
                PACKAGING / "upload-filters.json",
            )
            for outer_name in ("sources.zip", "targets.zip"):
                with zipfile.ZipFile(output / outer_name) as outer:
                    with zipfile.ZipFile(BytesIO(outer.read("workspace.zip"))) as filtered:
                        names = filtered.namelist()
                self.assertNotIn("workspace/archive/roundtrip-imports/previous.zip", names)
                self.assertIn("workspace/deployment/local/bin/create-chatgpt-upload-zips.sh", names)

    def test_filters_flat_runtime_root(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            runtime = root / "runtime.zip"
            runtime.write_bytes(self.repository("runtime", flat=True, extra={
                "test/report.json": b"test",
                "prod/state.json": b"prod",
                "production/README.md": b"production",
            }))
            target = root / "target.zip"; target.write_bytes(self.repository("runtime"))
            output = root / "result"
            MODULE.create_work_package(
                gap, [f"runtime={runtime}"], [f"runtime={target}"], output, self.filter_config(root)
            )
            with zipfile.ZipFile(output / "sources.zip") as outer:
                with zipfile.ZipFile(BytesIO(outer.read("runtime.zip"))) as filtered:
                    self.assertNotIn("test/report.json", filtered.namelist())
                    self.assertNotIn("prod/state.json", filtered.namelist())
                    self.assertNotIn("production/README.md", filtered.namelist())
            manifest = MODULE.validate_work_package(output)
            self.assertEqual(manifest["uploadFilters"][0]["rootForm"], "flat")

    def test_validation_rejects_reintroduced_filtered_path(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            gap = root / "gap.zip"; gap.write_bytes(self.gap())
            runtime = root / "runtime.zip"
            runtime.write_bytes(self.repository("runtime", extra={"runtime/test/report.json": b"test"}))
            target = root / "target.zip"; target.write_bytes(self.repository("runtime"))
            output = root / "result"
            MODULE.create_work_package(
                gap, [f"runtime={runtime}"], [f"runtime={target}"], output, self.filter_config(root)
            )
            with zipfile.ZipFile(output / "sources.zip") as outer:
                source_entries = {name: outer.read(name) for name in outer.namelist()}
            with zipfile.ZipFile(BytesIO(source_entries["runtime.zip"])) as filtered:
                runtime_entries = {name: filtered.read(name) for name in filtered.namelist()}
            runtime_entries["runtime/test/reintroduced.json"] = b"forbidden"
            source_entries["runtime.zip"] = MODULE.build_deterministic_zip(runtime_entries)
            (output / "sources.zip").write_bytes(MODULE.build_deterministic_zip(source_entries))
            with self.assertRaisesRegex(MODULE.WorkPackageError, "Negativprüfung fehlgeschlagen"):
                MODULE.validate_work_package(output)


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