195 lines
9.1 KiB
Python
195 lines
9.1 KiB
Python
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"]
|