#!/usr/bin/env python3

import hashlib
import importlib.machinery
import importlib.util
import json
import os
import subprocess
import tarfile
import tempfile
import unittest
from pathlib import Path


SCRIPT_DIR = Path(__file__).resolve().parent
PLATFORM_ROOT = SCRIPT_DIR.parent.parent
BUILDER = SCRIPT_DIR / "build-module-foundry-consumer-policy-artifact.mjs"
RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy"
PATCH_ID = "module-foundry-zone-v2-consumer-policy-unit-001"
EXPECTED_FILES = (
    "registry/data-product-consumer-policies.json",
    "scripts/validate-registry.mjs",
    "server/foundry-data-product-consumer.mjs",
    "server/foundry-data-product-consumer.test.mjs",
)


def load_runner():
    loader = importlib.machinery.SourceFileLoader(
        "nodedc_module_foundry_consumer_policy_artifact",
        str(RUNNER_PATH),
    )
    spec = importlib.util.spec_from_loader(loader.name, loader)
    module = importlib.util.module_from_spec(spec)
    loader.exec_module(module)
    return module


RUNNER = load_runner()


class ModuleFoundryConsumerPolicyArtifactTest(unittest.TestCase):
    def build(self, artifact_dir):
        environment = os.environ.copy()
        environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir)
        completed = subprocess.run(
            ["node", str(BUILDER), PATCH_ID],
            cwd=PLATFORM_ROOT,
            env=environment,
            check=True,
            capture_output=True,
            text=True,
        )
        return json.loads(completed.stdout)

    def test_builder_is_deterministic_exact_scope_and_runner_compatible(self):
        with tempfile.TemporaryDirectory(prefix="nodedc-foundry-policy-zone-v2-") as directory:
            root = Path(directory)
            first = self.build(root / "first")
            second = self.build(root / "second")
            first_artifact = Path(first["artifact"])
            second_artifact = Path(second["artifact"])

            self.assertEqual(first_artifact.read_bytes(), second_artifact.read_bytes())
            self.assertEqual(first["sha256"], hashlib.sha256(first_artifact.read_bytes()).hexdigest())
            self.assertEqual(tuple(first["files"]), EXPECTED_FILES)

            extracted = root / "loaded"
            extracted.mkdir()
            manifest, entries, payload = RUNNER.load_artifact(first_artifact, extracted)
            self.assertEqual(manifest["component"], "module-foundry")
            self.assertEqual(tuple(entries), EXPECTED_FILES)
            self.assertEqual(RUNNER.component_services("module-foundry", entries), ("nodedc-module-foundry",))

            registry = json.loads((payload / EXPECTED_FILES[0]).read_text(encoding="utf-8"))
            policies = [
                policy for policy in registry["policies"]
                if policy["dataProductId"] == "fleet.positions.current.v4"
                and policy["productVersion"] == "4.0.0"
            ]
            self.assertEqual(len(policies), 1)
            policy = policies[0]
            self.assertEqual(policy["id"], "map-moving-object-current-v4")
            self.assertEqual(policy["statusContract"]["attribute"], "signal_state")
            self.assertEqual(policy["statusContract"]["allowedValues"], ["active", "inactive"])
            self.assertEqual(policy["statusContract"]["freshness"], "none")
            self.assertIsNone(policy["staleAfterMs"])

            historical_zone_policies = [
                policy for policy in registry["policies"]
                if policy["dataProductId"] == "map.zones.current.v1"
                and policy["productVersion"] == "1.0.0"
            ]
            self.assertEqual(historical_zone_policies, [{
                "id": "map-zone-current-v1",
                "version": "1.0.0",
                "dataProductId": "map.zones.current.v1",
                "productVersion": "1.0.0",
                "freshness": "none",
                "staleAfterMs": None,
                "terminalStatuses": [],
                "removeMode": "canonical-tombstone-or-snapshot-rebase",
            }])

            zone_policies = [
                policy for policy in registry["policies"]
                if policy["dataProductId"] == "map.zones.current.v2"
                and policy["productVersion"] == "2.0.0"
            ]
            self.assertEqual(len(zone_policies), 1)
            zone_policy = zone_policies[0]
            self.assertEqual(zone_policy["id"], "map-zone-current-v2")
            self.assertEqual(zone_policy["version"], "2.0.0")
            self.assertEqual(zone_policy["freshness"], "none")
            self.assertIsNone(zone_policy["staleAfterMs"])
            self.assertEqual(zone_policy["consumerContract"]["semanticTypes"], ["map.zone"])
            self.assertEqual(zone_policy["consumerContract"]["fieldProjection"], [
                "display_name",
                "geometry_kind",
                "max_speed_kph",
                "schedule_timezone",
                "applies_to_couriers",
                "applies_to_kicksharing",
            ])
            self.assertEqual(zone_policy["consumerContract"]["geometryTypes"], ["Polygon", "MultiPolygon"])
            self.assertEqual(zone_policy["consumerContract"]["subjectIdentity"], "semantic-type+source-id")
            self.assertEqual(zone_policy["consumerContract"]["snapshotMode"], "atomic-replace")
            self.assertEqual(zone_policy["consumerContract"]["patchMode"], "atomic")
            self.assertFalse(any(
                word in json.dumps(zone_policy).lower()
                for word in ("provider", "tenant", "connection", "endpoint", "credential", "token", "secret", "authorization")
            ))

            profile_policies = [
                policy for policy in registry["policies"]
                if policy["dataProductId"] == "fleet.units.profile.current.v1"
                and policy["productVersion"] == "1.0.0"
            ]
            self.assertEqual(len(profile_policies), 1)
            profile_policy = profile_policies[0]
            self.assertEqual(profile_policy["id"], "map-moving-object-unit-profile-current-v1")
            self.assertEqual(profile_policy["freshness"], "none")
            self.assertIsNone(profile_policy["staleAfterMs"])
            self.assertEqual(profile_policy["terminalStatuses"], [])
            self.assertEqual(profile_policy["consumerContract"]["ontologyRevision"], "ontology.map.moving_object.v3")
            self.assertEqual(profile_policy["consumerContract"]["semanticTypes"], ["map.moving_object"])
            self.assertEqual(len(profile_policy["consumerContract"]["fieldProjection"]), 28)
            self.assertEqual(profile_policy["consumerContract"]["geometryTypes"], [])
            self.assertEqual(profile_policy["consumerContract"]["subjectIdentity"], "semantic-type+source-id")
            self.assertFalse(any(
                word in json.dumps(profile_policy).lower()
                for word in ("provider", "tenant", "endpoint", "credential", "token", "secret", "authorization")
            ))

            with tarfile.open(first_artifact, "r:gz") as archive:
                members = archive.getmembers()
                names = [member.name for member in members]
                regular_payloads = [
                    archive.extractfile(member).read()
                    for member in members
                    if member.isfile()
                ]
            self.assertFalse(any(Path(name).name.startswith("._") for name in names))
            self.assertFalse(any("/.git/" in name or "/node_modules/" in name or "/runtime-data/" in name for name in names))
            self.assertNotIn(b"-----BEGIN PRIVATE KEY-----", b"\n".join(regular_payloads))


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