feat(clean-release): complete and verify backend test suite (33 passing tests)

- Relocated and standardized tests for clean_release subsystem into __tests__ sub-packages.
- Implemented missing unit tests for preparation_service, audit_service, and stages.
- Enhanced API contract tests for candidate preparation and compliance reporting.
- Updated 023-clean-repo-enterprise coverage matrix with final verification results.
- Fixed relative import issues and model validation mismatches during test migration.
This commit is contained in:
2026-03-04 13:53:43 +03:00
parent 7194f6a4c4
commit 0894254b98
20 changed files with 2729 additions and 1209 deletions

View File

@@ -0,0 +1,157 @@
# [DEF:backend.tests.api.routes.test_clean_release_api:Module]
# @TIER: STANDARD
# @SEMANTICS: tests, api, clean-release, checks, reports
# @PURPOSE: Contract tests for clean release checks and reports endpoints.
# @LAYER: Domain
# @RELATION: TESTS -> backend.src.api.routes.clean_release
# @INVARIANT: API returns deterministic payload shapes for checks and reports.
from datetime import datetime, timezone
from fastapi.testclient import TestClient
from src.app import app
from src.dependencies import get_clean_release_repository
from src.models.clean_release import (
CleanProfilePolicy,
ProfileType,
ReleaseCandidate,
ReleaseCandidateStatus,
ResourceSourceEntry,
ResourceSourceRegistry,
ComplianceReport,
CheckFinalStatus,
)
from src.services.clean_release.repository import CleanReleaseRepository
def _repo_with_seed_data() -> CleanReleaseRepository:
repo = CleanReleaseRepository()
repo.save_candidate(
ReleaseCandidate(
candidate_id="2026.03.03-rc1",
version="2026.03.03",
profile=ProfileType.ENTERPRISE_CLEAN,
created_at=datetime.now(timezone.utc),
created_by="tester",
source_snapshot_ref="git:abc123",
status=ReleaseCandidateStatus.PREPARED,
)
)
repo.save_registry(
ResourceSourceRegistry(
registry_id="registry-internal-v1",
name="Internal",
entries=[
ResourceSourceEntry(
source_id="src-1",
host="repo.intra.company.local",
protocol="https",
purpose="artifact-repo",
enabled=True,
)
],
updated_at=datetime.now(timezone.utc),
updated_by="tester",
status="active",
)
)
repo.save_policy(
CleanProfilePolicy(
policy_id="policy-enterprise-clean-v1",
policy_version="1.0.0",
active=True,
prohibited_artifact_categories=["test-data"],
required_system_categories=["system-init"],
external_source_forbidden=True,
internal_source_registry_ref="registry-internal-v1",
effective_from=datetime.now(timezone.utc),
profile=ProfileType.ENTERPRISE_CLEAN,
)
)
return repo
def test_start_check_and_get_status_contract():
repo = _repo_with_seed_data()
app.dependency_overrides[get_clean_release_repository] = lambda: repo
try:
client = TestClient(app)
start = client.post(
"/api/clean-release/checks",
json={
"candidate_id": "2026.03.03-rc1",
"profile": "enterprise-clean",
"execution_mode": "tui",
"triggered_by": "tester",
},
)
assert start.status_code == 202
payload = start.json()
assert set(["check_run_id", "candidate_id", "status", "started_at"]).issubset(payload.keys())
check_run_id = payload["check_run_id"]
status_resp = client.get(f"/api/clean-release/checks/{check_run_id}")
assert status_resp.status_code == 200
status_payload = status_resp.json()
assert status_payload["check_run_id"] == check_run_id
assert "final_status" in status_payload
assert "checks" in status_payload
finally:
app.dependency_overrides.clear()
def test_get_report_not_found_returns_404():
repo = _repo_with_seed_data()
app.dependency_overrides[get_clean_release_repository] = lambda: repo
try:
client = TestClient(app)
resp = client.get("/api/clean-release/reports/unknown-report")
assert resp.status_code == 404
finally:
app.dependency_overrides.clear()
def test_get_report_success():
repo = _repo_with_seed_data()
report = ComplianceReport(
report_id="rep-1",
check_run_id="run-1",
candidate_id="2026.03.03-rc1",
generated_at=datetime.now(timezone.utc),
final_status=CheckFinalStatus.COMPLIANT,
operator_summary="all systems go",
structured_payload_ref="manifest-1",
violations_count=0,
blocking_violations_count=0
)
repo.save_report(report)
app.dependency_overrides[get_clean_release_repository] = lambda: repo
try:
client = TestClient(app)
resp = client.get("/api/clean-release/reports/rep-1")
assert resp.status_code == 200
assert resp.json()["report_id"] == "rep-1"
finally:
app.dependency_overrides.clear()
def test_prepare_candidate_api_success():
repo = _repo_with_seed_data()
app.dependency_overrides[get_clean_release_repository] = lambda: repo
try:
client = TestClient(app)
response = client.post(
"/api/clean-release/candidates/prepare",
json={
"candidate_id": "2026.03.03-rc1",
"artifacts": [{"path": "file1.txt", "category": "system-init", "reason": "core"}],
"sources": ["repo.intra.company.local"],
"operator_id": "operator-1",
},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "prepared"
assert "manifest_id" in data
finally:
app.dependency_overrides.clear()

View File

@@ -0,0 +1,97 @@
# [DEF:backend.tests.api.routes.test_clean_release_source_policy:Module]
# @TIER: STANDARD
# @SEMANTICS: tests, api, clean-release, source-policy
# @PURPOSE: Validate API behavior for source isolation violations in clean release preparation.
# @LAYER: Domain
# @RELATION: TESTS -> backend.src.api.routes.clean_release
# @INVARIANT: External endpoints must produce blocking violation entries.
from datetime import datetime, timezone
from fastapi.testclient import TestClient
from src.app import app
from src.dependencies import get_clean_release_repository
from src.models.clean_release import (
CleanProfilePolicy,
ProfileType,
ReleaseCandidate,
ReleaseCandidateStatus,
ResourceSourceEntry,
ResourceSourceRegistry,
)
from src.services.clean_release.repository import CleanReleaseRepository
def _repo_with_seed_data() -> CleanReleaseRepository:
repo = CleanReleaseRepository()
repo.save_candidate(
ReleaseCandidate(
candidate_id="2026.03.03-rc1",
version="2026.03.03",
profile=ProfileType.ENTERPRISE_CLEAN,
created_at=datetime.now(timezone.utc),
created_by="tester",
source_snapshot_ref="git:abc123",
status=ReleaseCandidateStatus.DRAFT,
)
)
repo.save_registry(
ResourceSourceRegistry(
registry_id="registry-internal-v1",
name="Internal",
entries=[
ResourceSourceEntry(
source_id="src-1",
host="repo.intra.company.local",
protocol="https",
purpose="artifact-repo",
enabled=True,
)
],
updated_at=datetime.now(timezone.utc),
updated_by="tester",
status="active",
)
)
repo.save_policy(
CleanProfilePolicy(
policy_id="policy-enterprise-clean-v1",
policy_version="1.0.0",
active=True,
prohibited_artifact_categories=["test-data"],
required_system_categories=["system-init"],
external_source_forbidden=True,
internal_source_registry_ref="registry-internal-v1",
effective_from=datetime.now(timezone.utc),
profile=ProfileType.ENTERPRISE_CLEAN,
)
)
return repo
def test_prepare_candidate_blocks_external_source():
repo = _repo_with_seed_data()
app.dependency_overrides[get_clean_release_repository] = lambda: repo
try:
client = TestClient(app)
response = client.post(
"/api/clean-release/candidates/prepare",
json={
"candidate_id": "2026.03.03-rc1",
"artifacts": [
{"path": "cfg/system.yaml", "category": "system-init", "reason": "required"}
],
"sources": ["repo.intra.company.local", "pypi.org"],
"operator_id": "release-manager",
},
)
assert response.status_code == 200
data = response.json()
assert data["status"] == "blocked"
assert any(v["category"] == "external-source" for v in data["violations"])
finally:
app.dependency_overrides.clear()

View File

@@ -182,6 +182,16 @@ async def get_task(
# @POST: Returns a list of log entries or raises 404.
# @RETURN: List[LogEntry] - List of log entries.
# @TIER: CRITICAL
# @TEST_CONTRACT get_task_logs_api ->
# {
# required_params: {task_id: str},
# optional_params: {level: str, source: str, search: str},
# invariants: ["returns 404 for non-existent task", "applies filters correctly"]
# }
# @TEST_FIXTURE valid_task_logs_request -> {"task_id": "test_1", "level": "INFO"}
# @TEST_EDGE task_not_found -> raises 404
# @TEST_EDGE invalid_limit -> Query(limit=0) returns 422
# @TEST_INVARIANT response_purity -> verifies: [valid_task_logs_request]
# @TEST_CONTRACT: TaskLogQueryInput -> List[LogEntry]
# @TEST_SCENARIO: existing_task_logs_filtered -> Returns filtered logs by level/source/search with pagination.
# @TEST_FIXTURE: valid_task_with_mixed_logs -> backend/tests/fixtures/task_logs/valid_task_with_mixed_logs.json