559 lines
21 KiB
Python
559 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import http.server
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
CONTROLLER = ROOT / "ops" / "gitea-deploy-broker"
|
|
|
|
|
|
@pytest.fixture
|
|
def controller(monkeypatch, tmp_path):
|
|
monkeypatch.syspath_prepend(str(CONTROLLER))
|
|
spec = importlib.util.spec_from_file_location(
|
|
"appops_self_upgrade_recovery_boundaries_test",
|
|
CONTROLLER / "status_server.py",
|
|
)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
monkeypatch.setattr(module, "WEBHOOK_INBOX_ROOT", tmp_path / "inbox")
|
|
module.WEBHOOK_INBOX_ROOT.mkdir(mode=0o700)
|
|
monkeypatch.setattr(module, "APPOPS_PROGRESS_ROOT", tmp_path / "progress")
|
|
monkeypatch.setattr(module, "SELF_UPGRADE_RECOVERY_ROOT", tmp_path / "recovery")
|
|
monkeypatch.setattr(
|
|
module,
|
|
"SELF_UPGRADE_RECOVERY_REQUEST_ROOT",
|
|
tmp_path / "recovery" / "requests",
|
|
)
|
|
return module
|
|
|
|
|
|
def _private_recovery_payload(module, recovery_id: str) -> dict[str, object]:
|
|
revision = "c" * 40
|
|
return {
|
|
"schema": module._SELF_UPGRADE_RECOVERY_SCHEMA,
|
|
"recoveryId": recovery_id,
|
|
"semanticSha256": module._webhook_semantic_digest(
|
|
9, "refs/heads/main", revision,
|
|
),
|
|
"repository": module.SELF_UPGRADE_REPOSITORY,
|
|
"revision": revision,
|
|
"originalRawSha256": "1" * 64,
|
|
"idempotencyKeySha256": "2" * 64,
|
|
"preflightSha256": "3" * 64,
|
|
"expectedCurrentRevision": "4" * 40,
|
|
"currentContainerId": "5" * 64,
|
|
"currentImageId": "sha256:" + "6" * 64,
|
|
"currentSourceTree": "7" * 40,
|
|
"runtimeFingerprint": "8" * 64,
|
|
"registryFingerprint": "9" * 64,
|
|
"sourceFingerprint": "a" * 64,
|
|
"stateSha256": "b" * 64,
|
|
"runtimeEnvironment": {"API_TOKEN": "recovery-secret-sentinel"},
|
|
"createdAt": 1,
|
|
}
|
|
|
|
|
|
def _recovery_delivery(module, *, secret: str) -> dict[str, object]:
|
|
return {
|
|
"schema": module._WEBHOOK_SELF_UPGRADE_RECOVERY_SCHEMA,
|
|
"semantic_sha256": "a" * 64,
|
|
"recovery_of_sha256": "b" * 64,
|
|
"recovery_id": str(uuid.UUID("12345678-1234-4234-8234-123456789abc")),
|
|
"idempotency_key_sha256": "c" * 64,
|
|
"preflight_sha256": "d" * 64,
|
|
"body_sha256": "e" * 64,
|
|
"delivery": "22345678-1234-4234-8234-123456789abc",
|
|
"gitea_repository_id": 9,
|
|
"ref": "refs/heads/main",
|
|
"repository": module.SELF_UPGRADE_REPOSITORY,
|
|
"revision": "f" * 40,
|
|
"source": "typed_self_upgrade_recovery",
|
|
"status": "awaiting_validation",
|
|
"validation": {
|
|
"last_result": "wait",
|
|
"required_contexts": ["Managed validation / full (push)"],
|
|
"observed": [],
|
|
"checked_at": 1,
|
|
"deadline_at": 2,
|
|
},
|
|
"audit": [
|
|
{"at": 1, "stage": "self_upgrade_recovery_requested"},
|
|
{"at": 2, "stage": "awaiting_validation"},
|
|
],
|
|
"received_at": 1,
|
|
"updated_at": 2,
|
|
# These fields must remain private even if a caller or record contains them.
|
|
"runtimeEnvironment": {"API_TOKEN": secret},
|
|
"privateCredential": secret,
|
|
}
|
|
|
|
|
|
def test_ordinary_deploy_strips_ambient_recovery_binding_and_rejects_bad_bindings(
|
|
controller, monkeypatch,
|
|
):
|
|
ambient_secret = "ambient-recovery-secret-sentinel"
|
|
monkeypatch.setenv("APPOPS_SELF_UPGRADE_RECOVERY_ID", ambient_secret)
|
|
monkeypatch.setenv("APPOPS_REGISTERED_LEGACY_REQUEST_ID", ambient_secret)
|
|
calls = []
|
|
|
|
def fake_run(command, **kwargs):
|
|
calls.append((command, kwargs))
|
|
return subprocess.CompletedProcess(command, 0)
|
|
|
|
monkeypatch.setattr(controller.subprocess, "run", fake_run)
|
|
ordinary = controller._run_registered_deploy("jens/quiz-monitor", "c" * 40)
|
|
assert ordinary.returncode == 0
|
|
command, kwargs = calls.pop()
|
|
assert command[-3:] == ["deploy", "jens/quiz-monitor", "c" * 40]
|
|
assert "APPOPS_SELF_UPGRADE_RECOVERY_ID" not in kwargs["env"]
|
|
assert "APPOPS_REGISTERED_LEGACY_REQUEST_ID" not in kwargs["env"]
|
|
assert ambient_secret not in json.dumps(kwargs["env"])
|
|
|
|
recovery_id = str(uuid.UUID("12345678-1234-4234-8234-123456789abc"))
|
|
authorized = controller._run_registered_deploy(
|
|
controller.SELF_UPGRADE_REPOSITORY,
|
|
"c" * 40,
|
|
self_upgrade_recovery_id=recovery_id,
|
|
)
|
|
assert authorized.returncode == 0
|
|
_, kwargs = calls.pop()
|
|
assert kwargs["env"]["APPOPS_SELF_UPGRADE_RECOVERY_ID"] == recovery_id
|
|
assert "APPOPS_REGISTERED_LEGACY_REQUEST_ID" not in kwargs["env"]
|
|
|
|
invalid_bindings = [
|
|
(controller.SELF_UPGRADE_REPOSITORY, "not-a-uuid", None),
|
|
("jens/quiz-monitor", recovery_id, None),
|
|
(controller.SELF_UPGRADE_REPOSITORY, recovery_id, recovery_id),
|
|
]
|
|
for repository, bad_recovery_id, registered_id in invalid_bindings:
|
|
with pytest.raises(ValueError):
|
|
controller._run_registered_deploy(
|
|
repository,
|
|
"c" * 40,
|
|
self_upgrade_recovery_id=bad_recovery_id,
|
|
registered_legacy_request_id=registered_id,
|
|
)
|
|
assert calls == []
|
|
|
|
|
|
def test_preflight_projects_runtime_identity_without_environment_values(
|
|
controller, monkeypatch,
|
|
):
|
|
project = {
|
|
"branch": "main",
|
|
"gitea_repository_id": 9,
|
|
"containers": ["Unraid-AppOps-Gateway"],
|
|
"compose_project": "unraid-appops-gateway",
|
|
"compose_files": ["docker-compose.yml"],
|
|
"environment": {
|
|
"APPOPS_DATA_PATH": controller.SELF_UPGRADE_DATA_PATH,
|
|
"APPOPS_PUBLISH_HOST": controller.SELF_UPGRADE_PUBLISH_HOST,
|
|
},
|
|
"verify_oci_revision": True,
|
|
"validation": {
|
|
"validation_gate_version": 1,
|
|
"required": True,
|
|
"contexts": ["Managed validation / full (push)"],
|
|
"timeout_seconds": 1800,
|
|
"poll_interval_seconds": 5,
|
|
"max_poll_interval_seconds": 30,
|
|
},
|
|
}
|
|
private_environment = {"API_TOKEN": "preflight-secret-sentinel"}
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_load_projects",
|
|
lambda: {controller.SELF_UPGRADE_REPOSITORY: project},
|
|
)
|
|
monkeypatch.setattr(controller, "_self_upgrade_protected_head", lambda _revision: True)
|
|
monkeypatch.setattr(controller, "_is_current_branch_head", lambda *_args: True)
|
|
monkeypatch.setattr(
|
|
controller, "_fetch_gitea_validation", lambda *_args: ("pass", []),
|
|
)
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_self_upgrade_runtime",
|
|
lambda: {
|
|
"revision": "4" * 40,
|
|
"imageId": "sha256:" + "5" * 64,
|
|
"containerId": "6" * 64,
|
|
"imageRef": controller.SELF_UPGRADE_PRODUCTION_IMAGE_REF,
|
|
"sourceTree": "7" * 40,
|
|
"runtimeFingerprint": "8" * 64,
|
|
"runtimeEnvironment": private_environment,
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
controller.shutil,
|
|
"disk_usage",
|
|
lambda _path: type("Usage", (), {"free": 50 * 1024**3})(),
|
|
)
|
|
monkeypatch.setattr(controller, "_self_upgrade_protected_head", lambda _revision: True)
|
|
monkeypatch.setattr(controller, "_is_current_branch_head", lambda *_args: True)
|
|
monkeypatch.setattr(
|
|
controller, "_fetch_gitea_validation", lambda *_args: ("pass", []),
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "_validation_policy",
|
|
lambda _repository, _project: {
|
|
"contexts": ["Managed validation / full (push)"],
|
|
},
|
|
)
|
|
monkeypatch.setattr(controller, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True)
|
|
monkeypatch.setattr(controller.Handler, "log_message", lambda *args: None)
|
|
|
|
result = controller._self_upgrade_preflight("c" * 40)
|
|
serialized = json.dumps(result)
|
|
assert result["status"] == "preflight_ok"
|
|
assert "runtimeEnvironment" not in result
|
|
assert "API_TOKEN" not in serialized
|
|
assert "preflight-secret-sentinel" not in serialized
|
|
|
|
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), controller.Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
body = json.dumps({
|
|
"repository": controller.SELF_UPGRADE_REPOSITORY,
|
|
"revision": "c" * 40,
|
|
"reason": "preflight boundary test",
|
|
"expectedCurrentRevision": None,
|
|
}).encode("utf-8")
|
|
request = urllib.request.Request(
|
|
f"http://127.0.0.1:{server.server_port}{controller.SELF_UPGRADE_PREFLIGHT_PATH}",
|
|
data=body,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
response_body = response.read()
|
|
assert response.status == 200
|
|
response_payload = json.loads(response_body)
|
|
serialized_response = response_body.decode("utf-8")
|
|
assert response_payload["status"] == "preflight_ok"
|
|
assert "runtimeEnvironment" not in response_payload
|
|
assert "API_TOKEN" not in serialized_response
|
|
assert "preflight-secret-sentinel" not in serialized_response
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def test_public_recovery_status_projects_away_private_environment(controller, monkeypatch):
|
|
secret = "public-status-secret-sentinel"
|
|
record = _recovery_delivery(controller, secret=secret)
|
|
monkeypatch.setattr(controller, "_self_upgrade_delivery_record", lambda _request_id: record)
|
|
monkeypatch.setattr(controller, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True)
|
|
monkeypatch.setattr(controller, "_read_inbox_record", lambda _path: None)
|
|
monkeypatch.setattr(controller.Handler, "log_message", lambda *args: None)
|
|
|
|
server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), controller.Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
request = urllib.request.Request(
|
|
f"http://127.0.0.1:{server.server_port}"
|
|
f"{controller.SELF_UPGRADE_STATUS_PREFIX}{'a' * 64}",
|
|
method="GET",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=5) as response:
|
|
status = response.status
|
|
body = response.read()
|
|
except urllib.error.HTTPError as error:
|
|
status = error.code
|
|
body = error.read()
|
|
assert status == 200, body.decode("utf-8", errors="replace")
|
|
view = json.loads(body)
|
|
assert view["repository"] == controller.SELF_UPGRADE_REPOSITORY
|
|
assert view["revision"] == "f" * 40
|
|
assert view["attemptType"] == "self-upgrade-recovery"
|
|
assert view["recoveryOf"] == "b" * 64
|
|
assert not {
|
|
"recovery_id", "recovery_of_sha256", "idempotency_key_sha256",
|
|
"preflight_sha256", "body_sha256", "recovery_hmac",
|
|
} & view.keys()
|
|
assert "runtimeEnvironment" not in view
|
|
assert "privateCredential" not in view
|
|
assert "recovery_hmac" not in view
|
|
assert "idempotency_key_sha256" not in view
|
|
assert secret.encode() not in body
|
|
assert "API_TOKEN" not in body.decode("utf-8")
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=5)
|
|
|
|
|
|
def test_private_recovery_journal_is_0600_and_hmac_tampering_fails_closed(
|
|
controller, monkeypatch, tmp_path,
|
|
):
|
|
if os.name == "nt" or os.geteuid() != 0:
|
|
pytest.skip("The controller recovery store requires root-owned POSIX files")
|
|
recovery_root = tmp_path / "recovery-store"
|
|
recovery_root.mkdir(mode=0o700)
|
|
recovery_root.chmod(0o700)
|
|
monkeypatch.setattr(controller, "SELF_UPGRADE_RECOVERY_ROOT", recovery_root)
|
|
secret = b"fixed-self-upgrade-recovery-test-key-0123456789"
|
|
monkeypatch.setattr(controller, "_self_upgrade_secret", lambda: secret)
|
|
recovery_id = str(uuid.UUID("12345678-1234-4234-8234-123456789abc"))
|
|
payload = _private_recovery_payload(controller, recovery_id)
|
|
|
|
controller._self_upgrade_write_private_journal(recovery_id, payload, secret)
|
|
marker = recovery_root / f"{recovery_id}.json"
|
|
assert stat.S_IMODE(marker.stat().st_mode) == 0o600
|
|
assert controller._self_upgrade_recovery_journal(recovery_id) == payload
|
|
|
|
envelope = json.loads(marker.read_text(encoding="utf-8"))
|
|
envelope["payload"]["runtimeEnvironment"]["API_TOKEN"] = "changed-secret"
|
|
marker.write_text(json.dumps(envelope, sort_keys=True, separators=(",", ":")) + "\n")
|
|
marker.chmod(0o600)
|
|
assert controller._self_upgrade_recovery_journal(recovery_id) is None
|
|
|
|
envelope = controller._self_upgrade_recovery_envelope(payload, secret)
|
|
marker.write_text(json.dumps(envelope, sort_keys=True, separators=(",", ":")) + "\n")
|
|
marker.chmod(0o644)
|
|
assert controller._self_upgrade_recovery_journal(recovery_id) is None
|
|
|
|
|
|
def test_recovery_request_is_immutable_and_idempotent_for_one_key(
|
|
controller, monkeypatch, tmp_path,
|
|
):
|
|
if os.name == "nt" or os.geteuid() != 0:
|
|
pytest.skip("The controller recovery store requires root-owned POSIX files")
|
|
|
|
recovery_root = tmp_path / "recovery-store"
|
|
recovery_root.mkdir(mode=0o700)
|
|
recovery_root.chmod(0o700)
|
|
request_root = recovery_root / "requests"
|
|
monkeypatch.setattr(controller, "SELF_UPGRADE_RECOVERY_ROOT", recovery_root)
|
|
monkeypatch.setattr(controller, "SELF_UPGRADE_RECOVERY_REQUEST_ROOT", request_root)
|
|
|
|
secret = b"fixed-self-upgrade-recovery-test-key-0123456789"
|
|
monkeypatch.setattr(controller, "_self_upgrade_secret", lambda: secret)
|
|
revision = "c" * 40
|
|
semantic = controller._webhook_semantic_digest(9, "refs/heads/main", revision)
|
|
original_path = tmp_path / "original-failed-delivery.json"
|
|
original_raw = b'{"immutable":"original failed delivery"}\n'
|
|
original_path.write_bytes(original_raw)
|
|
original = {"revision": revision}
|
|
original_info = (original_path, original_raw, original)
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_self_upgrade_original_failed_delivery",
|
|
lambda *_args: original_info,
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "_self_upgrade_delivery_revision", lambda _semantic: revision,
|
|
)
|
|
|
|
expected_preflight = "3" * 64
|
|
expected_current_revision = controller.SELF_UPGRADE_OLD_REVISION
|
|
expected_container_id = "5" * 64
|
|
expected_image_id = "sha256:" + "6" * 64
|
|
expected_source_tree = "7" * 40
|
|
expected_runtime_fingerprint = "8" * 64
|
|
private_environment = {"API_TOKEN": "journal-secret-sentinel"}
|
|
fresh = {
|
|
"status": "preflight_ok",
|
|
"preflightSha256": expected_preflight,
|
|
"currentRevision": expected_current_revision,
|
|
"currentContainerId": expected_container_id,
|
|
"currentImageId": expected_image_id,
|
|
"currentSourceTree": expected_source_tree,
|
|
"currentRuntimeFingerprint": expected_runtime_fingerprint,
|
|
}
|
|
bindings = {
|
|
"currentRevision": expected_current_revision,
|
|
"currentContainerId": expected_container_id,
|
|
"currentImageId": expected_image_id,
|
|
"currentSourceTree": expected_source_tree,
|
|
"runtimeFingerprint": expected_runtime_fingerprint,
|
|
"registryFingerprint": "9" * 64,
|
|
"sourceFingerprint": "a" * 64,
|
|
"stateSha256": "b" * 64,
|
|
"runtimeEnvironment": private_environment,
|
|
}
|
|
monkeypatch.setattr(controller, "_self_upgrade_preflight", lambda _revision: fresh)
|
|
monkeypatch.setattr(
|
|
controller, "_self_upgrade_recovery_current_bindings", lambda: bindings,
|
|
)
|
|
|
|
policy = {
|
|
"repository": controller.SELF_UPGRADE_REPOSITORY,
|
|
"validation_gate_version": 1,
|
|
"contexts": ["Managed validation / full (push)"],
|
|
"timeout_seconds": 1800,
|
|
"poll_interval_seconds": 5,
|
|
"max_poll_interval_seconds": 30,
|
|
}
|
|
project = {"slug": "unraid-appops-gateway"}
|
|
monkeypatch.setattr(
|
|
controller, "_load_projects",
|
|
lambda: {controller.SELF_UPGRADE_REPOSITORY: project},
|
|
)
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_validation_policy",
|
|
lambda _repository, _project: policy,
|
|
)
|
|
|
|
original_sha = controller.hashlib.sha256(original_raw).hexdigest()
|
|
idempotency_key_sha = "2" * 64
|
|
reason_sha = "4" * 64
|
|
first = controller._self_upgrade_recovery_request_status(
|
|
semantic,
|
|
original_sha,
|
|
idempotency_key_sha,
|
|
expected_preflight,
|
|
expected_current_revision,
|
|
reason_sha,
|
|
policy,
|
|
)
|
|
assert first == ("awaiting_validation", semantic)
|
|
assert original_path.read_bytes() == original_raw
|
|
|
|
def store_snapshot() -> dict[str, bytes]:
|
|
return {
|
|
path.relative_to(recovery_root).as_posix(): path.read_bytes()
|
|
for path in sorted(recovery_root.rglob("*.json"))
|
|
}
|
|
|
|
accepted_bytes = store_snapshot()
|
|
assert accepted_bytes
|
|
assert stat.S_IMODE(recovery_root.stat().st_mode) == 0o700
|
|
assert stat.S_IMODE(request_root.stat().st_mode) == 0o700
|
|
assert all(
|
|
stat.S_IMODE(path.stat().st_mode) == 0o600
|
|
for path in recovery_root.rglob("*.json")
|
|
)
|
|
# The secret is intentionally present only in the root-owned HMAC journal.
|
|
recovery_id = str(uuid.UUID(bytes=bytes.fromhex(idempotency_key_sha)[:16], version=4))
|
|
journal_bytes = accepted_bytes[f"{recovery_id}.json"]
|
|
assert b"journal-secret-sentinel" in journal_bytes
|
|
assert b"journal-secret-sentinel" not in accepted_bytes[
|
|
f"{semantic}.delivery.json"
|
|
]
|
|
assert b"journal-secret-sentinel" not in accepted_bytes[
|
|
f"requests/{idempotency_key_sha}.json"
|
|
]
|
|
|
|
same_key = controller._self_upgrade_recovery_request_status(
|
|
semantic,
|
|
original_sha,
|
|
idempotency_key_sha,
|
|
expected_preflight,
|
|
expected_current_revision,
|
|
reason_sha,
|
|
policy,
|
|
)
|
|
assert same_key == ("awaiting_validation", semantic)
|
|
assert store_snapshot() == accepted_bytes
|
|
assert original_path.read_bytes() == original_raw
|
|
|
|
different_key = controller._self_upgrade_recovery_request_status(
|
|
semantic,
|
|
original_sha,
|
|
"d" * 64,
|
|
expected_preflight,
|
|
expected_current_revision,
|
|
reason_sha,
|
|
policy,
|
|
)
|
|
assert different_key == ("conflict", semantic)
|
|
assert store_snapshot() == accepted_bytes
|
|
assert original_path.read_bytes() == original_raw
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"malformed",
|
|
[b"{", b"[]", b'{"duplicate":1,"duplicate":2}'],
|
|
)
|
|
def test_legacy_json_parser_rejects_malformed_or_duplicate_records(
|
|
controller, monkeypatch, malformed,
|
|
):
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_self_upgrade_read_legacy_cache_file",
|
|
lambda *_args, **_kwargs: malformed,
|
|
)
|
|
with pytest.raises((OSError, ValueError)):
|
|
controller._self_upgrade_read_legacy_cache_json(
|
|
Path("/mnt/cache/appdata/gitea-deploy-runners/state/webhook-inbox")
|
|
/ ("a" * 64 + ".json"),
|
|
maximum_bytes=65536,
|
|
allowed_modes=frozenset({0o600}),
|
|
)
|
|
|
|
|
|
def test_original_failed_delivery_rejects_modified_legacy_record(controller, monkeypatch):
|
|
revision = "c" * 40
|
|
semantic = controller._webhook_semantic_digest(9, "refs/heads/main", revision)
|
|
policy = {
|
|
"repository": controller.SELF_UPGRADE_REPOSITORY,
|
|
"validation_gate_version": 1,
|
|
"contexts": ["Managed validation / full (push)"],
|
|
"timeout_seconds": 1800,
|
|
"poll_interval_seconds": 5,
|
|
"max_poll_interval_seconds": 30,
|
|
}
|
|
validation = controller._validation_snapshot(policy, 1)
|
|
validation["last_result"] = "pass"
|
|
validation["observed"] = [{
|
|
"context": "Managed validation / full (push)",
|
|
"status": "success",
|
|
}]
|
|
record = {
|
|
"schema": controller._WEBHOOK_SCHEMA,
|
|
"semantic_sha256": semantic,
|
|
"gitea_repository_id": 9,
|
|
"ref": "refs/heads/main",
|
|
"repository": controller.SELF_UPGRADE_REPOSITORY,
|
|
"revision": revision,
|
|
"status": "failed",
|
|
"reason": "deployment_failed",
|
|
"source": "gitea_webhook",
|
|
"body_sha256": "1" * 64,
|
|
"validation": validation,
|
|
"audit": [
|
|
{"stage": "validation_passed"},
|
|
{"stage": "deployment_started"},
|
|
{"stage": "deployment_failed", "reason": "deployment_failed"},
|
|
],
|
|
}
|
|
raw = json.dumps(record, sort_keys=True, separators=(",", ":")).encode()
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_self_upgrade_read_legacy_cache_json",
|
|
lambda *_args, **_kwargs: (raw, record),
|
|
)
|
|
assert controller._self_upgrade_original_failed_delivery(
|
|
semantic, revision, policy,
|
|
) == (controller.WEBHOOK_INBOX_ROOT / f"{semantic}.json", raw, record)
|
|
|
|
tampered = {**record, "revision": "d" * 40}
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"_self_upgrade_read_legacy_cache_json",
|
|
lambda *_args, **_kwargs: (raw, tampered),
|
|
)
|
|
assert controller._self_upgrade_original_failed_delivery(
|
|
semantic, revision, policy,
|
|
) is None
|