from __future__ import annotations

import json
import tempfile
import unittest
from pathlib import Path
from uuid import uuid4

from packages.connector_kit.gitea import GiteaSettings
from packages.connector_kit.knowledge import (
    DocumentNotFound,
    InMemoryKnowledgeIndex,
    PathEscapeError,
    RegisteredSource,
)
from packages.connector_kit.redaction import safe_content
from packages.hub_contracts.dataplane import AliasRecord
from scripts.generate_sbom import generate
from scripts.security_scan import _BEARER, scan_repository
from services.control_api.catalog.models import CatalogState
from services.control_api.catalog.planner import build_plan
from services.control_api.contextforge.models import GatewayDesiredState, GatewaySnapshot


class SecurityPlanTests(unittest.TestCase):
    def test_bearer_secret_scan_does_not_cross_configuration_lines(self) -> None:
        self.assertIsNone(_BEARER.search("HUB_CONTEXTFORGE_AUTH_SCHEME: Bearer\nNEXT_KEY=value"))
        self.assertIsNotNone(_BEARER.search("Authorization: Bearer abcdefghijklmnopqrstuvwxyz1234"))

    def test_static_security_scan_proves_compose_invariants_and_no_live_secret(self) -> None:
        report = scan_repository()
        self.assertEqual(report["status"], "pass")
        self.assertEqual(report["compose"]["violations"], [])
        self.assertEqual(report["secrets"]["findings"], [])
        self.assertTrue(report["invariants"]["denyByDefault"])

    def test_sbom_is_deterministic_and_contains_locked_dependencies_and_images(self) -> None:
        report = generate()
        self.assertEqual(report["bomFormat"], "CycloneDX")
        self.assertEqual(report["metadata"]["component"]["name"], "itworx-mcp-hub")
        self.assertTrue(report["metadata"]["sourceCommit"])
        components = report["components"]
        self.assertTrue(any(item["name"] == "fastapi" for item in components))
        self.assertTrue(
            any(
                item["type"] == "container" and "@sha256:" in item["version"] for item in components
            )
        )

    def test_prompt_injection_is_data_and_secret_content_is_rejected(self) -> None:
        fixture = json.loads(
            (
                Path(__file__).resolve().parents[1]
                / "tests/fixtures/security/wp160-untrusted-inputs.json"
            ).read_text(encoding="utf-8")
        )
        content = fixture["promptInjection"]
        self.assertEqual(safe_content(content), content)
        self.assertIsNone(safe_content(fixture["secretLike"]))

    def test_registered_paths_and_document_uris_cannot_escape(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            source = RegisteredSource(
                source_id=uuid4(),
                project_id=uuid4(),
                project_slug="ludarium",
                source_ref="docs",
                source_type="documentation",
                root=Path(directory),
            )
            with self.assertRaises(PathEscapeError):
                source.canonical_path("../secrets")
        with self.assertRaises(DocumentNotFound):
            InMemoryKnowledgeIndex().get_document(uuid4(), "itworx-docs://ludarium/docs/%2e%2e")

    def test_ssrf_and_token_substitution_boundaries_are_rejected(self) -> None:
        with self.assertRaises(ValueError):
            GiteaSettings(
                base_url="https://169.254.169.254", token_reference="docker-" + "secret://gitea"
            )
        with self.assertRaises(ValueError):
            GiteaSettings(
                base_url="https://localhost", token_reference="docker-" + "secret://gitea"
            )
        with self.assertRaises(ValueError):
            AliasRecord(
                project_slug="ludarium",
                environment="prod",
                alias="/mcp/ludarium/prod",
                upstream_url="https://127.0.0.1:4444/mcp",
            )

    def test_gateway_drift_is_reported_and_unknown_gateway_is_not_deleted(self) -> None:
        desired = GatewayDesiredState(slug="knowledge", url="https://knowledge.example/mcp")
        unknown = GatewaySnapshot(
            slug="operator-owned",
            url="https://operator.example/mcp",
            description="outside catalog",
            transport="STREAMABLEHTTP",
            enabled=True,
            status="ready",
        )
        state = CatalogState(
            documents=(),
            gateways=(desired,),
            desired_hash="0" * 64,
            source_files=("catalog/project.yaml",),
        )
        plan = build_plan(state, (unknown,), set())
        self.assertEqual(plan.actions[0].action, "create")
        self.assertEqual(plan.drift[0].kind, "unknown_gateway")
        self.assertNotIn("delete", {action.action for action in plan.actions})

    def test_json_scan_output_is_serializable_without_credentials(self) -> None:
        payload = json.dumps(scan_repository(), sort_keys=True)
        self.assertNotIn("BEGIN PRIVATE KEY", payload)
        self.assertNotIn("Authorization: Bearer", payload)


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