from __future__ import annotations

import importlib.util
import json
import sys
import types
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]


@pytest.fixture
def controller(monkeypatch):
    path = ROOT / "ops/gitea-deploy-broker/status_server.py"
    spec = importlib.util.spec_from_file_location("hub_candidate_delivery_test", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    monkeypatch.setattr(module, "_gitea_validation_settings",
                        lambda: ("https://gitea.example.com/api/v1", b"x" * 40))
    return module


def test_candidate_api_reads_only_fixed_hub_paths(controller, monkeypatch):
    opened = []

    def run(argv, **kwargs):
        opened.append((argv, kwargs))
        assert argv == [sys.executable, str(controller.ROOT / "hub_candidate_transport.py"), "json"]
        assert kwargs["timeout"] == 20
        assert kwargs["pass_fds"] == ()
        assert kwargs["stderr"] == controller.subprocess.DEVNULL
        assert set(kwargs["env"]) == {"PATH", "PYTHONDONTWRITEBYTECODE", "PYTHONIOENCODING"}
        return types.SimpleNamespace(returncode=0, stdout=b'{"workflows":[]}')

    monkeypatch.setattr(controller.subprocess, "run", run)
    prefix = "/repos/jens/itworx_mcp_hub"
    assert controller._hub_candidate_api_json(prefix + "/actions/workflows", query={}) == {"workflows": []}
    payload = json.loads(opened[0][1]["input"])
    assert payload["url"] == "https://gitea.example.com/api/v1" + prefix + "/actions/workflows"
    assert payload["outputFd"] is None
    assert payload["headers"]["Authorization"] == "token " + "x" * 40
    assert all("x" * 40 not in item for item in opened[0][0])
    for path, query in (
        ("https://other.invalid/api/v1", {}),
        (prefix + "/../other/actions/workflows", {}),
        (prefix + "/actions/runs", {"event": "pull_request"}),
        (prefix + "/actions/runs/1/jobs", {"page": "11", "limit": "100"}),
        (prefix + "/actions/runs/1/artifacts", {"name": "other-artifact"}),
    ):
        with pytest.raises(controller._HubCandidateTransportError):
            controller._hub_candidate_api_json(path, query=query)
    assert len(opened) == 1


@pytest.mark.parametrize("raw", [b'{"id":1,"id":2}', b"x" * (2 * 1024 * 1024 + 1)])
def test_candidate_metadata_is_bounded_and_rejects_duplicate_keys(controller, monkeypatch, raw):
    monkeypatch.setattr(controller.subprocess, "run", lambda *_args, **_kwargs:
                        types.SimpleNamespace(returncode=0, stdout=raw))
    with pytest.raises(controller._HubCandidateTransportError):
        controller._hub_candidate_api_json("/repos/jens/itworx_mcp_hub/branches/main", query={})


def test_candidate_transport_deadline_is_parent_enforced(controller, monkeypatch):
    def blocked(argv, **kwargs):
        assert kwargs["timeout"] == 20
        raise controller.subprocess.TimeoutExpired(argv, kwargs["timeout"], output=b"private")

    monkeypatch.setattr(controller.subprocess, "run", blocked)
    with pytest.raises(controller._HubCandidateTransportError, match="read_unavailable") as error:
        controller._hub_candidate_api_json("/repos/jens/itworx_mcp_hub/branches/main", query={})
    assert "private" not in str(error.value)


def test_artifact_transport_writes_only_parent_owned_fd(controller, monkeypatch, tmp_path):
    import os
    controller.STATE_ROOT = tmp_path
    parent = tmp_path / "mcp-hub-release-imports"
    parent.mkdir(mode=0o700)
    destination = parent / "candidate.zip"

    def run(argv, **kwargs):
        assert argv[-1] == "zip"
        assert kwargs["timeout"] == 1800
        payload = json.loads(kwargs["input"])
        assert set(payload) == {"url", "headers", "maxBytes", "outputFd"}
        assert str(destination) not in kwargs["input"].decode()
        assert kwargs["pass_fds"] == (payload["outputFd"],)
        os.write(payload["outputFd"], b"zip data")
        return types.SimpleNamespace(returncode=0, stdout=b'{"bytes":8}\n')

    monkeypatch.setattr(controller.subprocess, "run", run)
    controller._hub_candidate_download_artifact(123, destination, 1024)
    assert destination.read_bytes() == b"zip data"
    assert destination.stat().st_mode & 0o777 == 0o600


@pytest.mark.parametrize("receipt", [b'{"bytes":9}', b'{"bytes":true}', b'{"bytes":0}'])
def test_artifact_receipt_must_match_actual_file(controller, monkeypatch, tmp_path, receipt):
    controller.STATE_ROOT = tmp_path
    parent = tmp_path / "mcp-hub-release-imports"
    parent.mkdir(mode=0o700)
    monkeypatch.setattr(controller.subprocess, "run", lambda *_args, **_kwargs:
                        types.SimpleNamespace(returncode=0, stdout=receipt))
    with pytest.raises(controller._HubCandidateTransportError):
        controller._hub_candidate_download_artifact(123, parent / "candidate.zip", 1024)


def _delivery(controller, monkeypatch, tmp_path, outcome):
    record = {"repository": controller.CANONICAL_HUB_REPOSITORY, "revision": "a" * 40,
              "status": "validation_passed", "audit": []}
    marker = tmp_path / "inbox.json"
    monkeypatch.setattr(controller, "_atomic_inbox_update",
                        lambda _marker, item: marker.write_text(json.dumps(item)))
    monkeypatch.setattr(controller, "_is_current_branch_head", lambda *_args: True)
    calls = []

    def prepare(*args, **kwargs):
        calls.append((args, kwargs))
        return {"revision": record["revision"], "status": outcome,
                "candidateSha256": "b" * 64}

    monkeypatch.setitem(sys.modules, "hub_candidate_import", types.SimpleNamespace(
        prepare_candidate=prepare, CandidateRejected=type("CandidateRejected", (ValueError,), {}),
        CandidateDeferred=type("CandidateDeferred", (RuntimeError,), {})))
    return marker, record, calls


@pytest.mark.parametrize("outcome", ["awaiting_candidate", "pending_review"])
def test_candidate_wait_never_marks_deployment_started_and_throttles(controller, monkeypatch, tmp_path, outcome):
    marker, record, calls = _delivery(controller, monkeypatch, tmp_path, outcome)
    assert not controller._prepare_hub_candidate_delivery(marker, record, {})
    assert record["status"] == "validation_passed"
    assert "deployment_started" not in [entry["stage"] for entry in record["audit"]]
    assert not controller._prepare_hub_candidate_delivery(marker, record, {})
    assert len(calls) == 1


def test_candidate_ready_is_hash_bound_before_deployment(controller, monkeypatch, tmp_path):
    marker, record, calls = _delivery(controller, monkeypatch, tmp_path, "ready")
    assert controller._prepare_hub_candidate_delivery(marker, record, {})
    assert record["status"] == "validation_passed"
    assert record["candidate_preparation"]["candidate_sha256"] == "b" * 64
    assert len(calls) == 1


def test_candidate_wait_expires_without_production_failure(controller, monkeypatch, tmp_path):
    marker, record, calls = _delivery(controller, monkeypatch, tmp_path, "awaiting_candidate")
    monkeypatch.setattr(controller.time, "time", lambda: 12000)
    record["candidate_preparation"] = {"started_at": 1, "deadline_at": 10801,
                                       "next_check_at": 2, "status": "awaiting_candidate"}
    assert not controller._prepare_hub_candidate_delivery(marker, record, {})
    assert record["status"] == "validation_timeout"
    assert record["reason"] == "hub_candidate_timeout"
    assert calls == []


def test_candidate_preparation_does_not_import_superseded_head(controller, monkeypatch, tmp_path):
    marker, record, calls = _delivery(controller, monkeypatch, tmp_path, "ready")
    monkeypatch.setattr(controller, "_is_current_branch_head", lambda *_args: False)
    assert not controller._prepare_hub_candidate_delivery(marker, record, {})
    assert record["status"] == "superseded"
    assert calls == []


def test_candidate_does_not_deploy_when_head_advances_during_import(controller, monkeypatch, tmp_path):
    marker, record, calls = _delivery(controller, monkeypatch, tmp_path, "ready")
    heads = iter((True, False))
    monkeypatch.setattr(controller, "_is_current_branch_head", lambda *_args: next(heads))
    assert not controller._prepare_hub_candidate_delivery(marker, record, {})
    assert len(calls) == 1
    assert record["status"] == "superseded"
    assert "candidate_sha256" not in record["candidate_preparation"]
    assert "deployment_started" not in [entry["stage"] for entry in record["audit"]]


def test_candidate_import_cannot_overrun_the_delivery_deadline(controller, monkeypatch, tmp_path):
    marker, record, calls = _delivery(controller, monkeypatch, tmp_path, "ready")
    clock = [1]
    monkeypatch.setattr(controller.time, "time", lambda: clock[0])
    prepare = sys.modules["hub_candidate_import"].prepare_candidate

    def slow(*args, **kwargs):
        result = prepare(*args, **kwargs)
        clock[0] = 10802
        return result

    monkeypatch.setattr(sys.modules["hub_candidate_import"], "prepare_candidate", slow)
    assert not controller._prepare_hub_candidate_delivery(marker, record, {})
    assert len(calls) == 1
    assert record["status"] == "validation_timeout"
    assert record["reason"] == "hub_candidate_timeout"
    assert "candidate_sha256" not in record["candidate_preparation"]