from __future__ import annotations

import dataclasses
import json
import sys
import tempfile
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
sys.path.insert(0, str(ROOT / "deployments/offline/runtime"))

from serviceification_d3_engine import CatalogPublisherKey, CatalogPublisherTrust, DeployError
from standalone_apprelease_publisher import prepare_upgrade_publisher


def trust(key_id: str, epoch: int = 4) -> CatalogPublisherTrust:
    return CatalogPublisherTrust("io.nexuscrm", epoch, True, (CatalogPublisherKey(
        key_id, "ed25519", "JAdozj6oo0RdiBhe2XZEsixSwOivaNfxlqageIHHmi0=",
        "2026-01-01T00:00:00Z", "2036-01-01T00:00:00Z",
    ),))


class StandalonePublisherTest(unittest.TestCase):
    def setUp(self) -> None:
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.root = Path(self.directory.name)
        self.current = trust("old")
        self.target = trust("new", 5)
        self.calls = []
        self.config = {
            "idempotencyKey": "test2-apprelease-fixed",
            "releaseImport": {"publisher": {
                "publisherId": self.target.publisher_id,
                "keys": [key.config_value() for key in self.target.keys],
            }},
        }

    def execute(self, argv) -> None:
        self.calls.append(tuple(argv))
        request = json.loads((self.root / "publisher-trust-config.json").read_text())
        self.assertEqual(request["expectedOwnershipEpoch"], "4")
        self.assertEqual(request["expectedActiveKeys"], [key.config_value() for key in self.current.keys])
        self.assertEqual(request["targetActiveKeys"], [key.config_value() for key in self.target.keys])
        result = {
            "schema": "nexuscrm.catalog-publisher-trust-transition-result/v1",
            "operationId": request["operationId"], "publisherId": "io.nexuscrm",
            "ownershipEpoch": "5", "previousTrustDigest": self.current.digest(),
            "activeTrustDigest": self.target.digest(),
        }
        (self.root / "publisher-trust-result.json").write_text(json.dumps(result))
        self.current = self.target

    def apply(self, **overrides) -> bool:
        inputs = dict(
            envv={}, catalog_config=self.config,
            catalog_image="catalog@sha256:" + "a" * 64,
            database_env=self.root / "database.env", evidence=self.root,
            execute=self.execute, read_publisher=lambda _: self.current,
            registered=lambda *_: False,
        )
        inputs.update(overrides)
        return prepare_upgrade_publisher(**inputs)

    def test_rotation_uses_existing_cli_and_exact_epoch_then_readback(self):
        self.assertTrue(self.apply())
        self.assertEqual(len(self.calls), 1)
        self.assertIn("publisher-trust", self.calls[0])
        self.assertIn("catalog@sha256:" + "a" * 64, self.calls[0])
        self.assertEqual((self.root / "publisher-trust-config.json").stat().st_mode & 0o777, 0o400)

    def test_same_key_upgrade_does_not_rotate(self):
        self.current = self.target
        self.assertFalse(self.apply())
        self.assertEqual(self.calls, [])

    def test_registered_release_does_not_reenable_historical_key(self):
        def forbidden_read(_):
            self.fail("an admitted release must not recheck publisher trust")
        self.assertFalse(self.apply(registered=lambda *_: True, read_publisher=forbidden_read))
        self.assertEqual(self.calls, [])

    def test_revoked_publisher_is_not_reactivated(self):
        self.current = dataclasses.replace(self.current, active=False)
        with self.assertRaisesRegex(DeployError, "active publisher"):
            self.apply()
        self.assertEqual(self.calls, [])

    def test_transition_failure_stops_before_import(self):
        def fail(_):
            raise DeployError("CAS conflict")
        with self.assertRaisesRegex(DeployError, "CAS conflict"):
            self.apply(execute=fail)

    def test_ack_loss_retry_reads_committed_target_without_second_transition(self):
        def lost(argv):
            self.execute(argv)
            raise DeployError("response lost after commit")
        with self.assertRaisesRegex(DeployError, "response lost"):
            self.apply(execute=lost)
        self.assertFalse(self.apply())
        self.assertEqual(len(self.calls), 1)

    def test_mismatched_receipt_and_readback_fail_closed(self):
        for corrupt_receipt in (True, False):
            with self.subTest(corrupt_receipt=corrupt_receipt):
                (self.root / "publisher-trust-config.json").unlink(missing_ok=True)
                self.current = trust("old")
                def corrupt(argv):
                    self.execute(argv)
                    if corrupt_receipt:
                        (self.root / "publisher-trust-result.json").write_text("{}")
                    else:
                        self.current = trust("old", 6)
                with self.assertRaisesRegex(DeployError, "disagrees"):
                    self.apply(execute=corrupt)

    def test_both_controllers_prepare_trust_before_catalog_import(self):
        for environment in ("test1", "test2"):
            source = (ROOT / f"scripts/manage_{environment}_apprelease.py").read_text()
            self.assertIn("if before is not None:\n            prepare_upgrade_publisher(", source)
            self.assertLess(source.index("prepare_upgrade_publisher(\n"), source.index('catalog_output_path ='))


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