from __future__ import annotations

import importlib.util
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path


SCRIPTS = Path(__file__).resolve().parent
REPO = SCRIPTS.parent
HELPER = SCRIPTS / "public-runtime-upload-root.ps1"
MANIFEST_PROBE = SCRIPTS / "validate-public-runtime-upload-manifest.py"
OFFLINE_QUIESCENCE_PROBE = SCRIPTS / "validate-public-runtime-offline-quiescence.py"
DATABASE_IDENTITY = SCRIPTS / "public_runtime_database_identity.py"
INITIALIZER = SCRIPTS / "initialize-public-runtime-upload-root.ps1"
INITIALIZER_WORKER = SCRIPTS / "initialize-public-runtime-upload-root.py"
PROCESS_PROBE = SCRIPTS / "probe-public-runtime-upload-root.py"
START = SCRIPTS / "start-public-runtime.ps1"
BOOT = SCRIPTS / "boot-public-runtime.ps1"
WATCH = SCRIPTS / "watch-public-runtime.ps1"
INSTALL = SCRIPTS / "install-public-runtime-task.ps1"
REGISTER = SCRIPTS / "register-boot-task.ps1"


def _read(path: Path) -> str:
    return path.read_text(encoding="utf-8")


def _load(path: Path, name: str):
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise AssertionError(f"could not import {path}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _run_windows_powershell(script: Path) -> subprocess.CompletedProcess[str]:
    powershell = shutil.which("powershell.exe")
    if powershell is None:
        raise unittest.SkipTest("Windows PowerShell 5.1 is unavailable")
    return subprocess.run(
        [
            powershell,
            "-NoLogo",
            "-NoProfile",
            "-NonInteractive",
            "-ExecutionPolicy",
            "Bypass",
            "-File",
            str(script),
        ],
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=30,
    )


class PublicRuntimeUploadReleaseSafetyTest(unittest.TestCase):
    @unittest.skipUnless(shutil.which("git.exe"), "git.exe is unavailable")
    def test_drive_and_unc_share_roots_are_rejected_before_path_trimming(self) -> None:
        with tempfile.TemporaryDirectory(
            prefix="vignette-upload-root-boundary-"
        ) as raw:
            root = Path(raw)
            source = root / "source"
            source.mkdir()
            subprocess.run(
                [shutil.which("git.exe") or "git.exe", "-C", str(source), "init"],
                check=True,
                capture_output=True,
            )
            harness = root / "root-boundary.ps1"
            quoted_helper = str(HELPER).replace("'", "''")
            quoted_source = str(source).replace("'", "''")
            harness.write_text(
                f"""$ErrorActionPreference = 'Stop'
. '{quoted_helper}'
function Assert-Rejected([string]$Candidate) {{
  $rejected = $false
  try {{
    Resolve-PublicRuntimeUploadRoot `
      -SourceRoot '{quoted_source}' `
      -UploadRoot $Candidate `
      -CreateIfMissing | Out-Null
  }} catch {{
    $rejected = $true
  }}
  if (-not $rejected) {{ throw "filesystem root accepted: $Candidate" }}
}}
Assert-Rejected 'D:\\'
Assert-Rejected '\\\\server\\share\\'
""",
                encoding="utf-8-sig",
            )
            result = _run_windows_powershell(harness)
            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout={result.stdout}\nstderr={result.stderr}",
            )

    def test_manifest_and_freeze_paths_must_stay_outside_public_and_source_roots(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory(
            prefix="vignette-upload-private-boundary-"
        ) as raw:
            root = Path(raw)
            source = root / "source"
            upload = root / "uploads"
            private = root / "private"
            source.mkdir()
            upload.mkdir()
            private.mkdir()
            inside_upload = upload / "manifest.json"
            inside_source = source / "manifest.json"
            outside = private / "manifest.json"
            for path in (inside_upload, inside_source, outside):
                path.write_text("{}", encoding="utf-8")
            harness = root / "private-boundary.ps1"

            def quote(value: Path) -> str:
                return str(value).replace("'", "''")

            harness.write_text(
                f"""$ErrorActionPreference = 'Stop'
. '{quote(HELPER)}'
function Assert-Rejected([string]$Candidate) {{
  $rejected = $false
  try {{
    Resolve-PublicRuntimePrivateStatePath `
      -SourceRoot '{quote(source)}' `
      -UploadRoot '{quote(upload)}' `
      -StatePath $Candidate `
      -RequireFile | Out-Null
  }} catch {{
    $rejected = $true
  }}
  if (-not $rejected) {{ throw "public state path accepted: $Candidate" }}
}}
Assert-Rejected '{quote(inside_upload)}'
Assert-Rejected '{quote(inside_source)}'
$accepted = Resolve-PublicRuntimePrivateStatePath `
  -SourceRoot '{quote(source)}' `
  -UploadRoot '{quote(upload)}' `
  -StatePath '{quote(outside)}' `
  -RequireFile
if ($accepted -cne (Resolve-Path -LiteralPath '{quote(outside)}').Path) {{
  throw 'external private manifest was not preserved'
}}
""",
                encoding="utf-8-sig",
            )
            result = _run_windows_powershell(harness)
            self.assertEqual(
                result.returncode,
                0,
                msg=f"stdout={result.stdout}\nstderr={result.stderr}",
            )

    def test_every_consumer_requires_one_pinned_validated_manifest(self) -> None:
        for path in (START, BOOT, WATCH, INSTALL, REGISTER):
            source = _read(path)
            with self.subTest(path=path.name):
                self.assertIn("[string]$UserUploadManifestPath", source)
                self.assertIn("[string]$ExpectedUserUploadManifestSha256", source)
                self.assertIn("[string]$UserUploadWriteFreezePath", source)
                self.assertIn("Test-PublicRuntimeUploadManifest", source)
                self.assertIn("validate-public-runtime-upload-manifest.py", source)

        for path in (START, INSTALL, REGISTER):
            source = _read(path)
            call = source.index(
                "$resolvedUserUploadDir = Resolve-PublicRuntimeUploadRoot"
            )
            call_window = source[call : call + 360]
            with self.subTest(no_implicit_create=path.name):
                self.assertNotIn("-CreateIfMissing", call_window)

        hidden = _read(SCRIPTS / "watch-public-runtime-hidden.vbs")
        for marker in (
            "-UserUploadManifestPath",
            "-ExpectedUserUploadManifestSha256",
            "-UserUploadWriteFreezePath",
        ):
            self.assertIn(marker, hidden)

        validator = _read(MANIFEST_PROBE)
        offline_validator = _read(OFFLINE_QUIESCENCE_PROBE)
        database_identity = _read(DATABASE_IDENTITY)
        self.assertIn("SELECT avatar_url", validator)
        self.assertIn("validate_current_avatar_references", validator)
        self.assertIn("verify_preserved_objects=True", validator)
        self.assertIn("reject_unbound_extras=False", validator)
        self.assertIn('"current_reference_set_sha256"', validator)
        for field in (
            "preserved_decode_valid_count",
            "preserved_decode_invalid_count",
            "required_decode_invalid_object_count",
            "required_decode_invalid_reference_count",
        ):
            with self.subTest(validator_field=field):
                self.assertIn(f'"{field}"', validator)
                self.assertIn(f'"{field}"', offline_validator)
        for field in (
            "current_decode_invalid_object_count",
            "current_decode_invalid_reference_count",
        ):
            with self.subTest(current_validator_field=field):
                self.assertIn(f'"{field}"', validator)

        start_contract = _read(START)
        self.assertIn("$offlinePreservedDecodeCounts.ValidCount", start_contract)
        self.assertIn("$manifestPreservedDecodeCounts.ValidCount", start_contract)
        self.assertIn("$offlineRequiredDecodeCounts.ObjectCount", start_contract)
        self.assertIn("$manifestRequiredDecodeCounts.ObjectCount", start_contract)
        self.assertIn("$manifestCurrentDecodeCounts.ObjectCount", start_contract)

    def test_initializer_is_the_only_explicit_root_creator_and_is_copy_only(
        self,
    ) -> None:
        self.assertTrue(INITIALIZER.is_file())
        self.assertTrue(INITIALIZER_WORKER.is_file())
        self.assertTrue(MANIFEST_PROBE.is_file())
        wrapper = _read(INITIALIZER)
        worker = _read(INITIALIZER_WORKER)
        self.assertIn("-CreateIfMissing", wrapper)
        self.assertIn("ExpectedReferenceCount", wrapper)
        self.assertIn("ExpectedPreservedObjectCount", wrapper)
        self.assertIn("ExpectedPreservedTotalSizeBytes", wrapper)
        self.assertIn("ExpectedPreservedInventorySha256", wrapper)
        self.assertIn("scan_preserved_inventory(source_roots)", worker)
        self.assertGreaterEqual(worker.count("assert_expected_preserved_inventory("), 3)
        self.assertIn("FileMode]::CreateNew", wrapper)
        self.assertIn('open("xb")', worker)
        self.assertNotIn("shutil.move", worker)
        self.assertNotIn("os.replace", worker)
        self.assertNotIn(
            'relative_path"', worker[worker.index("def privacy_safe_manifest") :]
        )

    def test_process_proof_is_bound_to_the_unique_loopback_listener_pid(self) -> None:
        module = _load(PROCESS_PROBE, "public_runtime_listener_probe")
        expected_root = r"C:\stable-uploads"
        expected_cwd = r"D:\release\apps\api"
        expected_manifest = r"C:\private\upload-manifest.json"
        expected_manifest_sha256 = "a" * 64
        expected_freeze = r"C:\private\upload-freeze.json"
        expected_database_target_sha256 = "b" * 64
        argv = [
            "python.exe",
            "-m",
            "uvicorn",
            "app.main:app",
            "--host",
            "127.0.0.1",
            "--port",
            "8001",
            "--workers",
            "1",
        ]
        snapshots = [
            {
                "pid": 10,
                "argv": argv,
                "cwd": expected_cwd,
                "environment": {
                    "USER_UPLOAD_DIR": expected_root,
                    "USER_UPLOAD_MANIFEST_REQUIRED": "true",
                    "USER_UPLOAD_MANIFEST_PATH": expected_manifest,
                    "USER_UPLOAD_MANIFEST_SHA256": expected_manifest_sha256,
                    "USER_UPLOAD_WRITE_FREEZE_PATH": expected_freeze,
                    "PUBLIC_RUNTIME_DB_TARGET_SHA256": (
                        expected_database_target_sha256
                    ),
                },
            },
            {
                "pid": 20,
                "argv": argv,
                "cwd": expected_cwd,
                "environment": {
                    "USER_UPLOAD_DIR": expected_root,
                    "USER_UPLOAD_MANIFEST_REQUIRED": "true",
                    "USER_UPLOAD_MANIFEST_PATH": expected_manifest,
                    "USER_UPLOAD_MANIFEST_SHA256": expected_manifest_sha256,
                    "USER_UPLOAD_WRITE_FREEZE_PATH": expected_freeze,
                    "PUBLIC_RUNTIME_DB_TARGET_SHA256": (
                        expected_database_target_sha256
                    ),
                },
            },
        ]
        passed = module.evaluate_listener_binding(
            listener_pids=[10],
            process_snapshots=snapshots,
            expected_root=expected_root,
            expected_api_cwd=expected_cwd,
            expected_manifest_path=expected_manifest,
            expected_manifest_sha256=expected_manifest_sha256,
            expected_write_freeze_path=expected_freeze,
            expected_database_target_sha256=expected_database_target_sha256,
            api_port=8001,
        )
        self.assertEqual(0, passed[0])
        self.assertEqual(10, passed[1]["pid"])

        decoy_only = module.evaluate_listener_binding(
            listener_pids=[30],
            process_snapshots=snapshots,
            expected_root=expected_root,
            expected_api_cwd=expected_cwd,
            expected_manifest_path=expected_manifest,
            expected_manifest_sha256=expected_manifest_sha256,
            expected_write_freeze_path=expected_freeze,
            expected_database_target_sha256=expected_database_target_sha256,
            api_port=8001,
        )
        self.assertNotEqual(0, decoy_only[0])
        self.assertEqual("listener_process_unavailable", decoy_only[1]["reason"])

        ambiguous = module.evaluate_listener_binding(
            listener_pids=[10, 20],
            process_snapshots=snapshots,
            expected_root=expected_root,
            expected_api_cwd=expected_cwd,
            expected_manifest_path=expected_manifest,
            expected_manifest_sha256=expected_manifest_sha256,
            expected_write_freeze_path=expected_freeze,
            expected_database_target_sha256=expected_database_target_sha256,
            api_port=8001,
        )
        self.assertNotEqual(0, ambiguous[0])
        self.assertEqual("api_listener_count_mismatch", ambiguous[1]["reason"])

    def test_listener_identity_call_chain_pins_receipt_and_database_target(
        self,
    ) -> None:
        contract = _read(HELPER)
        start = _read(START)
        validator = _read(MANIFEST_PROBE)
        database_identity = _read(DATABASE_IDENTITY)
        for expected in (
            '"--expected-manifest-path"',
            '"--expected-manifest-sha256"',
            '"--expected-write-freeze-path"',
            '"--expected-database-target-sha256"',
            "ListenerPid = $listenerPid",
        ):
            with self.subTest(source="contract", expected=expected):
                self.assertIn(expected, contract)

        prior_capture = start[
            start.index(
                "$priorApiListenerProof = Test-PublicRuntimeApiUploadRoot"
            ) : start.index("$priorCloudflaredProcesses = @(")
        ]
        self.assertIn("$priorApiListenerProof.ListenerPid", prior_capture)
        self.assertNotIn("Get-UvicornProcessesByPort", prior_capture)
        self.assertIn(
            "$env:PUBLIC_RUNTIME_DB_TARGET_SHA256 = $expectedDatabaseTargetSha256",
            start,
        )
        self.assertIn("connected_database_target_sha256", validator)
        self.assertIn("current_database()", database_identity)
        self.assertIn("current_user::text", database_identity)
        self.assertIn("inet_server_addr()", database_identity)
        self.assertIn('"database_target_sha256"', validator)

    def test_fresh_cutover_requires_drained_freeze_and_restores_it(self) -> None:
        source = _read(START)
        for expected in (
            "Assert-PublicUploadWriteFreezeReady",
            "Exit-PublicUploadWriteFreeze",
            "upload_write_freeze",
            '"USER_UPLOAD_MANIFEST_PATH"',
            '"USER_UPLOAD_MANIFEST_SHA256"',
            '"USER_UPLOAD_WRITE_FREEZE_PATH"',
            "Fresh public promotion requires a drained upload-write freeze",
            "Fresh public rollback did not restore upload-write availability",
        ):
            with self.subTest(expected=expected):
                self.assertIn(expected, source)

    def test_write_release_is_an_irreversible_cutover_boundary(self) -> None:
        source = _read(START)
        release_stage = source.index('$freshFailureStage = "upload_write_release"')
        release_call = source.index(
            "Exit-PublicUploadWriteFreeze `",
            release_stage,
        )
        no_rollback = source.index(
            "$freshNoRollback = $true",
            release_stage,
        )
        receipt_publish = source.index(
            '$freshFailureStage = "receipt_publish"',
            release_stage,
        )
        passed_receipt_write = source.index(
            "Write-Utf8TextAtomically `",
            receipt_publish,
        )
        committed = source.index(
            "$freshPromotionCommitted = $true",
            receipt_publish,
        )
        self.assertLess(no_rollback, release_call)
        self.assertLess(release_call, receipt_publish)
        self.assertLess(receipt_publish, passed_receipt_write)
        self.assertLess(passed_receipt_write, committed)

        trap = source[
            source.index("trap {") : source.index("if (!(Test-Path $Python))")
        ]
        rollback_guard = trap.index("-not $freshNoRollback")
        rollback_call = trap.index("Restore-PriorPublicRuntime `")
        self.assertLess(rollback_guard, rollback_call)

        # Exit deletes the sentinel before polling health. A timeout after that
        # deletion must not make the trap restore the prior upload root.
        exit_function = source[
            source.index("function Exit-PublicUploadWriteFreeze") : source.index(
                "function Assert-PublicUploadWritesAvailable"
            )
        ]
        self.assertLess(
            exit_function.index("[System.IO.File]::Delete($FreezePath)"),
            exit_function.index("Get-JsonHealth"),
        )

    def test_boot_and_watchdog_skip_mutation_during_valid_active_freeze(self) -> None:
        for path in (BOOT, WATCH):
            source = _read(path)
            with self.subTest(path=path.name):
                freeze_guard = source.index(
                    "promotion-in-progress: valid drained upload freeze is active"
                )
                self.assertIn("$promotionFreeze.active -eq $true", source)
                self.assertIn("$promotionFreeze.valid -eq $true", source)
                self.assertIn("[int]$promotionFreeze.in_flight -eq 0", source)
                self.assertIn(
                    "[string]$promotionFreeze.token_sha256 -ceq",
                    source,
                )
                if path.name == "boot-public-runtime.ps1":
                    mutation = source.index("& $startScript @startArgs")
                else:
                    mutation = source.index("& $startScript @startArgs")
                self.assertLess(freeze_guard, mutation)


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