"""Production licensing administration console tests."""

from __future__ import annotations

from contextlib import redirect_stderr, redirect_stdout
from io import StringIO
import json
import os
from pathlib import Path
import tempfile
import unittest

from licensing_shared.catalog import load_builtin_catalog
from licensing_server.admin_cli import (
    ADMIN_API_URL_ENVIRONMENT,
    ADMIN_TOKEN_FILE_ENVIRONMENT,
    main,
)


class _FakeResponse:
    def __init__(self, document: dict[str, object], status: int = 200) -> None:
        self.status = status
        self._body = json.dumps(document).encode("utf-8")

    def __enter__(self) -> "_FakeResponse":
        return self

    def __exit__(self, *_arguments: object) -> None:
        return None

    def read(self, maximum_bytes: int) -> bytes:
        return self._body[:maximum_bytes]


class _FakeUrlOpen:
    def __init__(self, *documents: dict[str, object]) -> None:
        self._responses = [_FakeResponse(value) for value in documents]
        self.calls: list[tuple[object, float]] = []

    def __call__(self, request: object, timeout: float) -> _FakeResponse:
        self.calls.append((request, timeout))
        if not self._responses:
            raise AssertionError("unexpected administration API call")
        return self._responses.pop(0)


class AdminCliTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.root = Path(self.temporary.name)
        self.token_path = self.root / "admin-token"
        self.token_path.write_text("test-admin-access-token", encoding="utf-8")
        self.token_path.chmod(0o600)
        self.environment = {
            ADMIN_API_URL_ENVIRONMENT: "https://licensing.example.test",
            ADMIN_TOKEN_FILE_ENVIRONMENT: str(self.token_path),
        }

    def tearDown(self) -> None:
        self.temporary.cleanup()

    def _run(
        self,
        arguments: list[str],
        requester: _FakeUrlOpen,
    ) -> tuple[int, str, str]:
        output = StringIO()
        error = StringIO()
        with redirect_stdout(output), redirect_stderr(error):
            result = main(
                arguments,
                environ=self.environment,
                urlopen=requester,
            )
        return result, output.getvalue(), error.getvalue()

    def test_offline_decision_previews_then_confirms_through_guarded_api(self) -> None:
        preview = {
            "id": "offline_request.test.console",
            "requestId": "offline_trial.test.console",
            "status": "pending",
            "decisionGeneration": 0,
        }
        preview_requester = _FakeUrlOpen(preview)
        result, output, error = self._run(
            [
                "offline",
                "approve",
                "offline_request.test.console",
                "--reason-code",
                "evaluation_request",
            ],
            preview_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertTrue(json.loads(output)["confirmationRequired"])
        self.assertEqual(len(preview_requester.calls), 1)

        confirmed_requester = _FakeUrlOpen(
            preview,
            {
                "requestId": "offline_trial.test.console",
                "status": "approved",
                "decisionGeneration": 1,
            },
        )
        result, output, error = self._run(
            [
                "offline",
                "approve",
                "offline_request.test.console",
                "--reason-code",
                "evaluation_request",
                "--note",
                "Reviewed evaluation",
                "--confirm",
            ],
            confirmed_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertEqual(json.loads(output)["status"], "approved")
        self.assertEqual(len(confirmed_requester.calls), 2)
        request = confirmed_requester.calls[1][0]
        self.assertEqual(request.get_method(), "POST")
        self.assertEqual(
            request.full_url,
            "https://licensing.example.test/v1/admin/offline-requests/"
            "offline_request.test.console/approve",
        )
        body = json.loads(request.data)
        self.assertEqual(body["expectedDecisionGeneration"], 0)
        self.assertEqual(body["reasonCode"], "evaluation_request")
        self.assertTrue(body["idempotencyKey"].startswith("admin-offline-approve-"))
        self.assertEqual(
            request.get_header("Authorization"),
            "Bearer test-admin-access-token",
        )

    def test_serial_batch_revocation_requires_reviewed_state_digest(self) -> None:
        state_digest = "ab" * 32
        preview = {
            "batchId": "batch.test.console",
            "batchStatus": "active",
            "activeSerialCount": 2,
            "redeemedSerialCount": 1,
            "wouldRevokeSerialCount": 2,
            "stateDigest": state_digest,
            "canExecute": True,
        }
        preview_requester = _FakeUrlOpen(preview)
        result, output, error = self._run(
            [
                "serial-batch",
                "revoke",
                "batch.test.console",
                "--reason-code",
                "serial_exposure",
                "--note",
                "Reviewed exposed export",
            ],
            preview_requester,
        )
        self.assertEqual(result, 0, error)
        document = json.loads(output)
        self.assertTrue(document["confirmationRequired"])
        self.assertEqual(document["requiredStateDigest"], state_digest)
        self.assertEqual(len(preview_requester.calls), 1)
        preview_call = preview_requester.calls[0][0]
        self.assertEqual(preview_call.get_method(), "POST")
        self.assertEqual(
            preview_call.full_url,
            "https://licensing.example.test/v1/admin/serial-batches/"
            "batch.test.console/revocation-preview",
        )

        missing_digest_requester = _FakeUrlOpen(preview)
        result, _output, error = self._run(
            [
                "serial-batch",
                "revoke",
                "batch.test.console",
                "--reason-code",
                "serial_exposure",
                "--confirm",
            ],
            missing_digest_requester,
        )
        self.assertEqual(result, 1)
        self.assertIn("--expected-state-digest is required", error)
        self.assertEqual(len(missing_digest_requester.calls), 1)

        confirmed_requester = _FakeUrlOpen(
            preview,
            {
                "batchId": "batch.test.console",
                "batchStatus": "revoked",
                "newlyRevokedSerialCount": 2,
                "executed": True,
                "idempotentReplay": False,
            },
        )
        result, output, error = self._run(
            [
                "serial-batch",
                "revoke",
                "batch.test.console",
                "--reason-code",
                "serial_exposure",
                "--note",
                "Reviewed exposed export",
                "--expected-state-digest",
                state_digest,
                "--confirm",
            ],
            confirmed_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertTrue(json.loads(output)["executed"])
        self.assertEqual(len(confirmed_requester.calls), 2)
        execute_call = confirmed_requester.calls[1][0]
        self.assertEqual(execute_call.get_method(), "POST")
        self.assertEqual(
            execute_call.full_url,
            "https://licensing.example.test/v1/admin/serial-batches/"
            "batch.test.console/revoke",
        )
        body = json.loads(execute_call.data)
        self.assertEqual(body["expectedStateDigest"], state_digest)
        self.assertEqual(body["reasonCode"], "serial_exposure")
        self.assertTrue(
            body["idempotencyKey"].startswith("admin-serial-batch-revoke-")
        )

    def test_privacy_erasure_requires_reviewed_state_and_scoped_https_api(self) -> None:
        state_digest = "cd" * 32
        preview = {
            "privacyRequestId": "privacy_request.test.console",
            "status": "pending",
            "canComplete": True,
            "blockingReasons": [],
            "stateDigest": state_digest,
        }
        preview_requester = _FakeUrlOpen(preview)
        result, output, error = self._run(
            [
                "privacy",
                "erase-account",
                "privacy_request.test.console",
                "--reason-code",
                "customer_request",
            ],
            preview_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertTrue(json.loads(output)["confirmationRequired"])
        self.assertEqual(
            json.loads(output)["requiredStateDigest"],
            state_digest,
        )
        self.assertEqual(len(preview_requester.calls), 1)
        self.assertEqual(
            preview_requester.calls[0][0].full_url,
            "https://licensing.example.test/v1/admin/privacy/deletion-requests/"
            "privacy_request.test.console/preview",
        )

        confirmed_requester = _FakeUrlOpen(
            preview,
            {
                "privacyRequestId": "privacy_request.test.console",
                "status": "completed",
                "executed": True,
                "customerRightsDeleted": False,
            },
        )
        result, output, error = self._run(
            [
                "privacy",
                "erase-account",
                "privacy_request.test.console",
                "--reason-code",
                "customer_request",
                "--note",
                "Authenticated customer request",
                "--expected-state-digest",
                state_digest,
                "--confirm",
            ],
            confirmed_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertTrue(json.loads(output)["executed"])
        self.assertEqual(len(confirmed_requester.calls), 2)
        execute_request = confirmed_requester.calls[1][0]
        self.assertEqual(
            execute_request.full_url,
            "https://licensing.example.test/v1/admin/privacy/deletion-requests/"
            "privacy_request.test.console/execute",
        )
        body = json.loads(execute_request.data)
        self.assertEqual(body["expectedStateDigest"], state_digest)
        self.assertEqual(body["reasonCode"], "customer_request")
        self.assertTrue(
            body["idempotencyKey"].startswith("admin-privacy-erase-")
        )
        self.assertEqual(
            execute_request.get_header("Authorization"),
            "Bearer test-admin-access-token",
        )

    def test_catalog_staging_and_publication_use_exact_reviewed_digests(self) -> None:
        catalog = load_builtin_catalog()
        candidate_path = self.root / "catalog-candidate.json"
        candidate_path.write_text(
            json.dumps(catalog.to_mapping()),
            encoding="utf-8",
        )
        stage_requester = _FakeUrlOpen(
            {
                "registered": True,
                "revision": catalog.revision,
                "catalogSha256": catalog.sha256(),
            }
        )
        result, output, error = self._run(
            [
                "catalog",
                "stage",
                "--input",
                str(candidate_path),
                "--reason-code",
                "planned_catalog_release",
                "--note",
                "Reviewed catalog",
            ],
            stage_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertTrue(json.loads(output)["registered"])
        stage_request = stage_requester.calls[0][0]
        self.assertEqual(stage_request.get_method(), "POST")
        self.assertEqual(
            stage_request.full_url,
            "https://licensing.example.test/v1/admin/catalog/releases/stage",
        )
        stage_body = json.loads(stage_request.data)
        self.assertEqual(stage_body["document"], catalog.to_mapping())
        self.assertTrue(stage_body["idempotencyKey"].startswith("admin-catalog-stage-"))

        state_digest = "cd" * 32
        status = {
            "releases": [
                {
                    "revision": catalog.revision,
                    "catalogSha256": catalog.sha256(),
                    "status": "staged",
                }
            ]
        }
        preview = {"canPublish": True, "stateDigest": state_digest}
        preview_requester = _FakeUrlOpen(status, preview)
        result, output, error = self._run(
            [
                "catalog",
                "publish",
                str(catalog.revision),
                "--reason-code",
                "planned_catalog_release",
            ],
            preview_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertEqual(json.loads(output)["requiredStateDigest"], state_digest)
        self.assertEqual(len(preview_requester.calls), 2)

        confirmed_requester = _FakeUrlOpen(
            status,
            preview,
            {"executed": True, "catalogReady": True},
        )
        result, output, error = self._run(
            [
                "catalog",
                "publish",
                str(catalog.revision),
                "--reason-code",
                "planned_catalog_release",
                "--expected-state-digest",
                state_digest,
                "--confirm",
            ],
            confirmed_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertTrue(json.loads(output)["executed"])
        self.assertEqual(len(confirmed_requester.calls), 3)
        publish_request = confirmed_requester.calls[2][0]
        self.assertEqual(publish_request.get_method(), "POST")
        publish_body = json.loads(publish_request.data)
        self.assertEqual(publish_body["expectedStateDigest"], state_digest)
        self.assertTrue(
            publish_body["idempotencyKey"].startswith("admin-catalog-publish-")
        )

    def test_surface_profile_and_trial_serial_commands_use_guarded_api_and_file(
        self,
    ) -> None:
        profile_requester = _FakeUrlOpen(
            {
                "profileId": "surface_profile.test.focused",
                "status": "active",
                "policy": {
                    "analysisIds": ["fft_spectrum"],
                    "mainTabIds": ["acquisition", "analysis"],
                },
            }
        )
        result, output, error = self._run(
            [
                "surfaces",
                "create-profile",
                "--name",
                "Focused",
                "--description",
                "Focused launch profile",
                "--analysis-id",
                "fft_spectrum",
                "--main-tab-id",
                "acquisition",
                "--main-tab-id",
                "analysis",
                "--reason",
                "Focused launch",
            ],
            profile_requester,
        )
        self.assertEqual(result, 0, error)
        self.assertEqual(
            json.loads(output)["profileId"],
            "surface_profile.test.focused",
        )
        profile_call = profile_requester.calls[0][0]
        self.assertEqual(
            profile_call.full_url,
            "https://licensing.example.test/v1/admin/product-surface-profiles",
        )
        profile_body = json.loads(profile_call.data)
        self.assertEqual(profile_body["analysisIds"], ["fft_spectrum"])
        self.assertEqual(profile_body["mainTabIds"], ["acquisition", "analysis"])

        export_path = self.root / "one-hour-trials.json"
        serial_requester = _FakeUrlOpen(
            {
                "batchId": "batch.test.one_hour",
                "skuId": "trial.full",
                "serials": ["APLN-TEST-ONE-HOUR"],
                "surfaceProfileId": "surface_profile.test.focused",
                "trialDurationHours": 1,
                "warning": "Shown once",
            }
        )
        result, output, error = self._run(
            [
                "serial-batch",
                "generate",
                "--sku-id",
                "trial.full",
                "--quantity",
                "1",
                "--reason",
                "One-hour test trial",
                "--surface-profile-id",
                "surface_profile.test.focused",
                "--trial-duration-hours",
                "1",
                "--output",
                str(export_path),
            ],
            serial_requester,
        )
        self.assertEqual(result, 0, error)
        summary = json.loads(output)
        self.assertEqual(summary["quantity"], 1)
        self.assertNotIn("serials", summary)
        self.assertEqual(
            json.loads(export_path.read_text(encoding="utf-8"))["serials"],
            ["APLN-TEST-ONE-HOUR"],
        )
        if os.name != "nt":
            self.assertEqual(export_path.stat().st_mode & 0o077, 0)
        serial_body = json.loads(serial_requester.calls[0][0].data)
        self.assertEqual(serial_body["trialDurationHours"], 1)
        self.assertEqual(
            serial_body["surfaceProfileId"],
            "surface_profile.test.focused",
        )

    def test_audit_export_is_owner_only_and_never_overwrites(self) -> None:
        export_path = self.root / "audit-export.json"
        requester = _FakeUrlOpen(
            {
                "schema": "apolon.licensing.audit-export",
                "schemaVersion": 1,
                "events": [],
                "sha256": "ab" * 32,
            }
        )
        result, output, error = self._run(
            ["audit", "export", "--output", str(export_path)],
            requester,
        )
        self.assertEqual(result, 0, error)
        self.assertEqual(json.loads(output)["status"], "exported")
        self.assertTrue(export_path.is_file())
        if os.name != "nt":
            self.assertEqual(export_path.stat().st_mode & 0o777, 0o600)

        duplicate = _FakeUrlOpen(
            {
                "schema": "apolon.licensing.audit-export",
                "schemaVersion": 1,
                "events": [],
                "sha256": "ab" * 32,
            }
        )
        result, _output, error = self._run(
            ["audit", "export", "--output", str(export_path)],
            duplicate,
        )
        self.assertEqual(result, 1)
        self.assertIn("already exists", error)

    def test_console_rejects_http_origins_and_non_owner_token_files(self) -> None:
        requester = _FakeUrlOpen({"requests": [], "limit": 50})
        self.environment[ADMIN_API_URL_ENVIRONMENT] = "http://licensing.example.test"
        result, _output, error = self._run(["offline", "list"], requester)
        self.assertEqual(result, 1)
        self.assertIn("HTTPS origin", error)
        self.assertEqual(requester.calls, [])

        if os.name != "nt":
            self.environment[ADMIN_API_URL_ENVIRONMENT] = (
                "https://licensing.example.test"
            )
            self.token_path.chmod(0o644)
            result, _output, error = self._run(["offline", "list"], requester)
            self.assertEqual(result, 1)
            self.assertIn("owner-only", error)
            self.assertEqual(requester.calls, [])


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