386 lines
13 KiB
Python
386 lines
13 KiB
Python
# #region Test.Api.CleanRelease [C:3] [TYPE Module] [SEMANTICS test,clean-release,api]
|
|
# @BRIEF Unit tests for clean release API routes — prepare, check, report.
|
|
# @RELATION BINDS_TO -> [CleanReleaseApi]
|
|
# @TEST_EDGE: active_policy_not_found -> 409
|
|
# @TEST_EDGE: candidate_not_found -> 409
|
|
# @TEST_EDGE: check_run_not_found -> 404
|
|
# @TEST_EDGE: report_not_found -> 404
|
|
# @TEST_EDGE: candidate_exists -> 409
|
|
|
|
import os
|
|
|
|
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:")
|
|
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-tests")
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
_src = str(Path(__file__).resolve().parent.parent.parent / "src")
|
|
if _src not in sys.path:
|
|
sys.path.insert(0, _src)
|
|
|
|
|
|
def _make_repo(overrides: dict | None = None) -> MagicMock:
|
|
"""Build a mock CleanReleaseRepository with bare-minimum defaults."""
|
|
repo = MagicMock()
|
|
repo.get_active_policy.return_value = MagicMock(id="policy-1", policy_id="policy-1")
|
|
repo.get_candidate.return_value = None
|
|
repo.get_check_run.return_value = None
|
|
repo.get_report.return_value = None
|
|
repo.get_manifests_by_candidate.return_value = []
|
|
repo.stage_runs = {}
|
|
repo.reports = {}
|
|
repo.check_runs = {}
|
|
repo.approval_decisions = []
|
|
repo.publication_records = []
|
|
repo.save_check_run.side_effect = lambda r: r
|
|
repo.save_manifest.side_effect = lambda m: m
|
|
repo.save_violation.side_effect = lambda v: v
|
|
repo.save_candidate.side_effect = lambda c: c
|
|
repo.save_artifact.side_effect = lambda a: a
|
|
if overrides:
|
|
for k, v in overrides.items():
|
|
setattr(repo, k, v)
|
|
return repo
|
|
|
|
|
|
def _make_client(repo: MagicMock | None = None) -> TestClient:
|
|
from src.api.routes.clean_release import router
|
|
from src.dependencies import get_clean_release_repository
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
|
|
if repo is None:
|
|
repo = _make_repo()
|
|
|
|
app.dependency_overrides[get_clean_release_repository] = lambda: repo
|
|
return TestClient(app)
|
|
|
|
|
|
# ── register_candidate_v2_endpoint ──
|
|
|
|
class TestRegisterCandidateV2:
|
|
"""POST /api/clean-release/candidates"""
|
|
|
|
def test_register_success(self):
|
|
repo = _make_repo()
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates", json={
|
|
"id": "cand-1",
|
|
"version": "1.0.0",
|
|
"source_snapshot_ref": "main",
|
|
"created_by": "test-user",
|
|
})
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["id"] == "cand-1"
|
|
assert data["version"] == "1.0.0"
|
|
assert data["status"] == "DRAFT"
|
|
repo.save_candidate.assert_called_once()
|
|
|
|
def test_register_duplicate_returns_409(self):
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = MagicMock(id="cand-1")
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates", json={
|
|
"id": "cand-1",
|
|
"version": "1.0.0",
|
|
"source_snapshot_ref": "main",
|
|
"created_by": "test-user",
|
|
})
|
|
assert resp.status_code == 409
|
|
assert "CANDIDATE_EXISTS" in resp.text
|
|
|
|
|
|
# ── import_candidate_artifacts_v2_endpoint ──
|
|
|
|
class TestImportArtifacts:
|
|
"""POST /api/clean-release/candidates/{candidate_id}/artifacts"""
|
|
|
|
def test_import_success(self):
|
|
candidate = MagicMock()
|
|
candidate.id = "cand-1"
|
|
candidate.status = "DRAFT"
|
|
candidate.transition_to = MagicMock()
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = candidate
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/cand-1/artifacts", json={
|
|
"artifacts": [{
|
|
"id": "art-1",
|
|
"path": "/tmp/file.tar.gz",
|
|
"sha256": "abc123",
|
|
"size": 1024,
|
|
}]
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "success"
|
|
repo.save_artifact.assert_called_once()
|
|
candidate.transition_to.assert_called_once()
|
|
|
|
def test_import_candidate_not_found(self):
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = None
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/cand-999/artifacts", json={
|
|
"artifacts": [{"id": "a1", "path": "/p", "sha256": "s", "size": 1}]
|
|
})
|
|
assert resp.status_code == 404
|
|
|
|
def test_import_empty_artifacts(self):
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = MagicMock(id="cand-1")
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/cand-1/artifacts", json={
|
|
"artifacts": []
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
def test_import_missing_field(self):
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = MagicMock(id="cand-1")
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/cand-1/artifacts", json={
|
|
"artifacts": [{"id": "a1", "path": "/p"}] # missing sha256, size
|
|
})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ── get_candidate_overview_v2_endpoint ──
|
|
|
|
class TestCandidateOverview:
|
|
"""GET /api/clean-release/candidates/{candidate_id}/overview"""
|
|
|
|
def test_overview_success(self):
|
|
candidate = MagicMock()
|
|
candidate.id = "cand-1"
|
|
candidate.version = "1.0.0"
|
|
candidate.source_snapshot_ref = "main"
|
|
candidate.status = "PREPARED"
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = candidate
|
|
client = _make_client(repo)
|
|
resp = client.get("/api/clean-release/candidates/cand-1/overview")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["candidate_id"] == "cand-1"
|
|
assert data["status"] == "PREPARED"
|
|
|
|
def test_overview_not_found(self):
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = None
|
|
client = _make_client(repo)
|
|
resp = client.get("/api/clean-release/candidates/cand-999/overview")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ── start_check ──
|
|
|
|
class TestStartCheck:
|
|
"""POST /api/clean-release/checks"""
|
|
|
|
def test_start_check_success(self):
|
|
candidate = MagicMock()
|
|
candidate.id = "cand-1"
|
|
candidate.status = "PREPARED"
|
|
repo = _make_repo()
|
|
repo.get_active_policy.return_value = MagicMock(id="policy-1", policy_id="p1")
|
|
repo.get_candidate.return_value = candidate
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/checks", json={
|
|
"candidate_id": "cand-1",
|
|
"profile": "enterprise-clean",
|
|
"execution_mode": "tui",
|
|
"triggered_by": "system",
|
|
})
|
|
assert resp.status_code == 202
|
|
data = resp.json()
|
|
assert data["status"] == "running"
|
|
assert data["candidate_id"] == "cand-1"
|
|
|
|
def test_start_check_no_policy(self):
|
|
repo = _make_repo()
|
|
repo.get_active_policy.return_value = None
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/checks", json={
|
|
"candidate_id": "cand-1",
|
|
})
|
|
assert resp.status_code == 409
|
|
assert "POLICY_NOT_FOUND" in resp.text
|
|
|
|
def test_start_check_no_candidate(self):
|
|
repo = _make_repo()
|
|
repo.get_active_policy.return_value = MagicMock(id="p1")
|
|
repo.get_candidate.return_value = None
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/checks", json={
|
|
"candidate_id": "cand-999",
|
|
})
|
|
assert resp.status_code == 409
|
|
assert "CANDIDATE_NOT_FOUND" in resp.text
|
|
|
|
|
|
# ── get_check_status ──
|
|
|
|
class TestGetCheckStatus:
|
|
"""GET /api/clean-release/checks/{check_run_id}"""
|
|
|
|
def test_check_found(self):
|
|
from datetime import datetime, timezone
|
|
run = MagicMock()
|
|
run.id = "check-1"
|
|
run.candidate_id = "cand-1"
|
|
run.final_status = "PASSED"
|
|
run.started_at = datetime.now(timezone.utc)
|
|
run.finished_at = datetime.now(timezone.utc)
|
|
repo = _make_repo()
|
|
repo.get_check_run.return_value = run
|
|
client = _make_client(repo)
|
|
resp = client.get("/api/clean-release/checks/check-1")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["check_run_id"] == "check-1"
|
|
|
|
def test_check_not_found(self):
|
|
repo = _make_repo()
|
|
repo.get_check_run.return_value = None
|
|
client = _make_client(repo)
|
|
resp = client.get("/api/clean-release/checks/check-999")
|
|
assert resp.status_code == 404
|
|
assert "CHECK_NOT_FOUND" in resp.text
|
|
|
|
|
|
# ── get_report ──
|
|
|
|
class TestGetReport:
|
|
"""GET /api/clean-release/reports/{report_id}"""
|
|
|
|
def test_report_found(self):
|
|
report = MagicMock()
|
|
report.id = "report-1"
|
|
report.run_id = "check-1"
|
|
report.candidate_id = "cand-1"
|
|
report.final_status = "PASSED"
|
|
repo = _make_repo()
|
|
repo.get_report.return_value = report
|
|
client = _make_client(repo)
|
|
resp = client.get("/api/clean-release/reports/report-1")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["report_id"] == "report-1"
|
|
|
|
def test_report_not_found(self):
|
|
repo = _make_repo()
|
|
repo.get_report.return_value = None
|
|
client = _make_client(repo)
|
|
resp = client.get("/api/clean-release/reports/report-999")
|
|
assert resp.status_code == 404
|
|
assert "REPORT_NOT_FOUND" in resp.text
|
|
|
|
|
|
# ── prepare_candidate_endpoint ──
|
|
|
|
@patch("src.api.routes.clean_release.prepare_candidate")
|
|
class TestPrepareCandidateEndpoint:
|
|
"""POST /api/clean-release/candidates/prepare"""
|
|
|
|
def test_prepare_success(self, mock_prepare):
|
|
mock_prepare.return_value = {"status": "prepared", "manifest_id": "m-1"}
|
|
repo = _make_repo()
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/prepare", json={
|
|
"candidate_id": "cand-1",
|
|
"artifacts": [],
|
|
"sources": ["src1"],
|
|
"operator_id": "op-1",
|
|
})
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "prepared"
|
|
|
|
def test_prepare_blocked(self, mock_prepare):
|
|
mock_prepare.return_value = {"status": "check_blocked"}
|
|
repo = _make_repo()
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/prepare", json={
|
|
"candidate_id": "cand-1",
|
|
"artifacts": [],
|
|
"sources": [],
|
|
"operator_id": "op-1",
|
|
})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "blocked"
|
|
|
|
def test_prepare_value_error(self, mock_prepare):
|
|
mock_prepare.side_effect = ValueError("Bad request")
|
|
repo = _make_repo()
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/prepare", json={
|
|
"candidate_id": "cand-1",
|
|
"artifacts": [],
|
|
"sources": [],
|
|
"operator_id": "op-1",
|
|
})
|
|
assert resp.status_code == 400
|
|
assert "CLEAN_PREPARATION_ERROR" in resp.text
|
|
|
|
|
|
# ── manifest build endpoint ──
|
|
|
|
class TestBuildManifest:
|
|
"""POST /api/clean-release/candidates/{candidate_id}/manifests"""
|
|
|
|
def test_build_manifest_success(self):
|
|
candidate = MagicMock()
|
|
candidate.id = "cand-1"
|
|
candidate.status = "PREPARED"
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = candidate
|
|
|
|
mock_manifest = MagicMock()
|
|
mock_manifest.id = "manifest-cand-1"
|
|
mock_manifest.candidate_id = "cand-1"
|
|
mock_manifest.manifest_version = 1
|
|
mock_manifest.manifest_digest = "digest-1"
|
|
mock_manifest.artifacts_digest = "art-digest-1"
|
|
mock_manifest.created_at = __import__("datetime").datetime.now()
|
|
mock_manifest.created_by = "system"
|
|
mock_manifest.source_snapshot_ref = "main"
|
|
mock_manifest.content_json = {}
|
|
|
|
with patch(
|
|
"src.services.clean_release.manifest_service.build_manifest_snapshot",
|
|
return_value=mock_manifest
|
|
):
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/cand-1/manifests", json={
|
|
"created_by": "system",
|
|
})
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["id"] == "manifest-cand-1"
|
|
assert data["manifest_version"] == 1
|
|
|
|
def test_build_manifest_value_error(self):
|
|
repo = _make_repo()
|
|
repo.get_candidate.return_value = MagicMock(id="cand-1")
|
|
|
|
with patch(
|
|
"src.services.clean_release.manifest_service.build_manifest_snapshot",
|
|
side_effect=ValueError("No artifacts")
|
|
):
|
|
client = _make_client(repo)
|
|
resp = client.post("/api/clean-release/candidates/cand-1/manifests", json={
|
|
"created_by": "system",
|
|
})
|
|
assert resp.status_code == 400
|
|
assert "MANIFEST_BUILD_ERROR" in resp.text
|
|
# #endregion Test.Api.CleanRelease
|