272 lines
12 KiB
Python
272 lines
12 KiB
Python
"""Offline tests for the D7-prep Gitea catalog publisher (no network)."""
|
|
# #region Test.Tooling.PublishCatalog [C:3] [TYPE Module] [SEMANTICS test,cli,gitea,catalog,publish]
|
|
# @BRIEF Offline fixtures for validate/path/fetch/put and the typed failure matrix.
|
|
# @RELATION VERIFIES -> [Tooling.PublishCatalog]
|
|
# @TEST_CONTRACT: validate_catalog -> parsed dict or PUBLISH_CATALOG_INVALID
|
|
# @TEST_EDGE: missing-env config -> exit 2 without network
|
|
# @TEST_INVARIANT Tooling.PublishCatalog: an invalid or nested-only catalog never reaches the network;
|
|
# create is POST (no sha) and update is PUT (sha) per the Gitea contract; auth,
|
|
# create-race and transport failures are typed. -> VERIFIED_BY: tests below
|
|
import base64
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src"))
|
|
|
|
from src.scripts.publish_catalog import ( # noqa: E402
|
|
GiteaTarget,
|
|
PublishError,
|
|
catalog_path,
|
|
fetch_existing_sha,
|
|
main,
|
|
publish_catalog_bytes,
|
|
validate_catalog,
|
|
)
|
|
|
|
_ENVELOPE = {
|
|
"baseline_set_id": "ss-prod-visual", "baseline_set_version": "1",
|
|
"release_id": "11111111-1111-4111-8111-111111111111", "baseline_family": "a" * 64,
|
|
"catalog_digest": "a" * 64,
|
|
"catalog_revision": {"catalog_revision_id": "rev-001", "publication": {"state": "published"}},
|
|
}
|
|
_RAW = json.dumps(_ENVELOPE).encode()
|
|
_NESTED_ONLY = json.dumps({
|
|
"catalog_revision": {"catalog_revision_id": "rev-001"},
|
|
"baseline_pin": {"baseline_set_id": "ss-prod-visual"},
|
|
}).encode()
|
|
_TARGET = GiteaTarget(base_url="http://gitea.test", repo="o/r", ref="main", token="t", path="catalogs/ss-prod-visual/v1.json")
|
|
|
|
|
|
# #region Test.Tooling.PublishCatalog.Fakes [C:2] [TYPE Class]
|
|
# @BRIEF Minimal httpx-client doubles capturing GET/POST/PUT calls for offline assertions.
|
|
class _FakeResponse:
|
|
def __init__(self, status_code: int, payload=None, text: str = "") -> None:
|
|
self.status_code = status_code
|
|
self._payload = payload if payload is not None else {}
|
|
self.text = text
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
|
|
class _FakeClient:
|
|
def __init__(self, get_response, put_response, post_response=None) -> None:
|
|
self.get_response = get_response
|
|
self.put_response = put_response
|
|
self.post_response = post_response or put_response
|
|
self.gets: list[tuple[str, dict]] = []
|
|
self.puts: list[tuple[str, dict]] = []
|
|
self.posts: list[tuple[str, dict]] = []
|
|
|
|
def get(self, url, params=None):
|
|
self.gets.append((url, params or {}))
|
|
return self.get_response
|
|
|
|
def put(self, url, json=None):
|
|
self.puts.append((url, json or {}))
|
|
return self.put_response
|
|
|
|
def post(self, url, json=None):
|
|
self.posts.append((url, json or {}))
|
|
return self.post_response
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
return False
|
|
# #endregion Test.Tooling.PublishCatalog.Fakes
|
|
|
|
|
|
# #region Test.Tooling.PublishCatalog.Validate [C:2] [TYPE Function]
|
|
def test_validate_catalog_accepts_envelope_and_rejects_nested_only_and_malformed():
|
|
assert validate_catalog(_RAW)["catalog_digest"] == "a" * 64
|
|
|
|
# A nested-only contract-refresh artifact must be rejected: the resolver cannot consume it.
|
|
try:
|
|
validate_catalog(_NESTED_ONLY)
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_CATALOG_INVALID" and "top-level identity" in exc.detail
|
|
else:
|
|
raise AssertionError("nested-only catalog must be rejected")
|
|
|
|
for bad in (
|
|
b"not-json", b"[]",
|
|
json.dumps({**_ENVELOPE, "release_id": ""}).encode(),
|
|
json.dumps({k: v for k, v in _ENVELOPE.items() if k != "catalog_revision"}).encode(),
|
|
):
|
|
try:
|
|
validate_catalog(bad)
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_CATALOG_INVALID"
|
|
else:
|
|
raise AssertionError(f"catalog {bad[:30]!r} must be rejected")
|
|
|
|
|
|
def test_catalog_path_renders_aliases_and_rejects_bad_template():
|
|
assert catalog_path("ss-prod-visual", "1", "catalogs/{baseline_set_id}/v{version}.json") == "catalogs/ss-prod-visual/v1.json"
|
|
assert catalog_path("ss-prod-visual", "1", "catalogs/{baseline_set}/v{baseline_set_version}.json") == "catalogs/ss-prod-visual/v1.json"
|
|
assert catalog_path("ss-prod-visual", "1", None) == "catalogs/ss-prod-visual/v1.json"
|
|
for bad_template in ("catalogs/{unknown}/v1.json", "catalogs/{baseline_set_id/v1.json"):
|
|
try:
|
|
catalog_path("ss-prod-visual", "1", bad_template)
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_CATALOG_PATH_INVALID"
|
|
else:
|
|
raise AssertionError(f"template {bad_template!r} must be rejected")
|
|
# #endregion Test.Tooling.PublishCatalog.Validate
|
|
|
|
|
|
# #region Test.Tooling.PublishCatalog.Publish [C:2] [TYPE Function]
|
|
def test_publish_creates_when_absent_and_updates_with_sha():
|
|
# Gitea contract: create is POST (no sha), update is PUT (sha required) — a PUT on a missing
|
|
# file is rejected 422 "[SHA]: Required" (verified live 2026-09-11).
|
|
create_client = _FakeClient(_FakeResponse(404), _FakeResponse(422, {"message": "[SHA]: Required"}), _FakeResponse(201, {"content": {"path": "x"}}))
|
|
assert fetch_existing_sha(create_client, _TARGET) is None
|
|
publish_catalog_bytes(create_client, _TARGET, _RAW, "create")
|
|
assert not create_client.puts
|
|
assert len(create_client.posts) == 1
|
|
assert "sha" not in create_client.posts[0][1]
|
|
assert base64.b64decode(create_client.posts[0][1]["content"]) == _RAW
|
|
|
|
update_client = _FakeClient(_FakeResponse(200, {"sha": "existing-sha"}), _FakeResponse(200, {}))
|
|
assert fetch_existing_sha(update_client, _TARGET) == "existing-sha"
|
|
publish_catalog_bytes(update_client, _TARGET, _RAW, "update")
|
|
assert not update_client.posts
|
|
assert update_client.puts[0][1]["sha"] == "existing-sha"
|
|
|
|
|
|
def test_publish_failures_are_typed():
|
|
auth_client = _FakeClient(_FakeResponse(401), _FakeResponse(200, {}))
|
|
try:
|
|
fetch_existing_sha(auth_client, _TARGET)
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_AUTH_FAILED"
|
|
else:
|
|
raise AssertionError("401 must be PUBLISH_AUTH_FAILED")
|
|
|
|
conflict_client = _FakeClient(_FakeResponse(200, {"sha": "s"}), _FakeResponse(409))
|
|
try:
|
|
publish_catalog_bytes(conflict_client, _TARGET, _RAW, "msg")
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_CONFLICT"
|
|
else:
|
|
raise AssertionError("409 must be PUBLISH_CONFLICT")
|
|
|
|
broken_client = _FakeClient(_FakeResponse(500, text="boom"), _FakeResponse(200, {}))
|
|
try:
|
|
fetch_existing_sha(broken_client, _TARGET)
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_UNAVAILABLE"
|
|
else:
|
|
raise AssertionError("500 must be PUBLISH_UNAVAILABLE")
|
|
|
|
# A create race (file appeared between GET 404 and POST) is a conflict, not unavailability.
|
|
race_client = _FakeClient(
|
|
_FakeResponse(404),
|
|
_FakeResponse(422, {"message": "[SHA]: Required"}, text="[SHA]: Required"),
|
|
_FakeResponse(422, {"message": "file already exists"}, text="file already exists"),
|
|
)
|
|
try:
|
|
publish_catalog_bytes(race_client, _TARGET, _RAW, "race")
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_CONFLICT"
|
|
else:
|
|
raise AssertionError("create race must be PUBLISH_CONFLICT")
|
|
|
|
# A POST auth failure is typed the same as a GET/PUT auth failure.
|
|
post_auth_client = _FakeClient(_FakeResponse(404), _FakeResponse(200, {}), _FakeResponse(401, text="bad token"))
|
|
try:
|
|
publish_catalog_bytes(post_auth_client, _TARGET, _RAW, "auth")
|
|
except PublishError as exc:
|
|
assert exc.code == "PUBLISH_AUTH_FAILED"
|
|
else:
|
|
raise AssertionError("POST 401 must be PUBLISH_AUTH_FAILED")
|
|
# #endregion Test.Tooling.PublishCatalog.Publish
|
|
|
|
|
|
# #region Test.Tooling.PublishCatalog.Main [C:2] [TYPE Function]
|
|
def test_main_missing_config_exits_2_without_network(monkeypatch, tmp_path):
|
|
catalog_file = tmp_path / "catalog.json"
|
|
catalog_file.write_bytes(_RAW)
|
|
monkeypatch.delenv("PUBLISHED_CATALOG_GITEA_URL", raising=False)
|
|
monkeypatch.delenv("PUBLISHED_CATALOG_GITEA_TOKEN", raising=False)
|
|
monkeypatch.delenv("PUBLISHED_CATALOG_REPO", raising=False)
|
|
monkeypatch.delenv("PUBLISHED_CATALOG_GITEA_REPO", raising=False)
|
|
assert main(["--catalog", str(catalog_file), "--baseline-set", "ss-prod-visual", "--version", "1"]) == 2
|
|
|
|
|
|
def test_main_publishes_end_to_end_with_fake_client(monkeypatch, tmp_path):
|
|
import src.scripts.publish_catalog as module
|
|
|
|
catalog_file = tmp_path / "catalog.json"
|
|
catalog_file.write_bytes(_RAW)
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_GITEA_URL", "http://gitea.test")
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_GITEA_TOKEN", "token-value")
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_REPO", "o/r")
|
|
seen: dict = {}
|
|
|
|
class _InstantClient(_FakeClient):
|
|
def __init__(self, **kwargs) -> None:
|
|
super().__init__(_FakeResponse(200, {"sha": "s"}), _FakeResponse(201, {"commit": {"id": "c1"}}))
|
|
seen["kwargs"] = kwargs
|
|
|
|
monkeypatch.setattr(module.httpx, "Client", lambda **kwargs: _InstantClient(**kwargs))
|
|
code = main(["--catalog", str(catalog_file), "--baseline-set", "ss-prod-visual", "--version", "1",
|
|
"--message", "publish test"])
|
|
assert code == 0
|
|
assert seen["kwargs"]["base_url"] == "http://gitea.test"
|
|
assert seen["kwargs"]["headers"] == {"Authorization": "token token-value"}
|
|
|
|
|
|
def test_main_creates_via_post_when_absent(monkeypatch, tmp_path):
|
|
import src.scripts.publish_catalog as module
|
|
|
|
catalog_file = tmp_path / "catalog.json"
|
|
catalog_file.write_bytes(_RAW)
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_GITEA_URL", "http://gitea.test")
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_GITEA_TOKEN", "token-value")
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_REPO", "o/r")
|
|
created: dict = {}
|
|
|
|
class _CreateClient(_FakeClient):
|
|
def __init__(self, **kwargs) -> None:
|
|
super().__init__(_FakeResponse(404), _FakeResponse(200, {}), _FakeResponse(201, {"commit": {"id": "c-new"}}))
|
|
|
|
def post(self, url, json=None):
|
|
created.setdefault("posts", []).append((url, json or {}))
|
|
return self.post_response
|
|
|
|
monkeypatch.setattr(module.httpx, "Client", lambda **kwargs: _CreateClient(**kwargs))
|
|
code = main(["--catalog", str(catalog_file), "--baseline-set", "ss-prod-visual", "--version", "1"])
|
|
assert code == 0
|
|
assert len(created["posts"]) == 1 and "sha" not in created["posts"][0][1]
|
|
|
|
|
|
def test_main_types_transport_errors(monkeypatch, tmp_path, capsys):
|
|
import httpx
|
|
|
|
import src.scripts.publish_catalog as module
|
|
|
|
catalog_file = tmp_path / "catalog.json"
|
|
catalog_file.write_bytes(_RAW)
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_GITEA_URL", "http://gitea.test")
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_GITEA_TOKEN", "token-value")
|
|
monkeypatch.setenv("PUBLISHED_CATALOG_REPO", "o/r")
|
|
|
|
class _BoomClient(_FakeClient):
|
|
def __init__(self, **kwargs) -> None:
|
|
super().__init__(_FakeResponse(404), _FakeResponse(201, {}))
|
|
|
|
def get(self, url, params=None):
|
|
raise httpx.ConnectError("boom")
|
|
|
|
monkeypatch.setattr(module.httpx, "Client", lambda **kwargs: _BoomClient(**kwargs))
|
|
code = main(["--catalog", str(catalog_file), "--baseline-set", "ss-prod-visual", "--version", "1"])
|
|
assert code == 1
|
|
assert "PUBLISH_UNAVAILABLE" in capsys.readouterr().out
|
|
# #endregion Test.Tooling.PublishCatalog.Main
|
|
# #endregion Test.Tooling.PublishCatalog
|