busya
f75c15dbc6
- 10 translate plugin test files (100% coverage on 12 modules) - assistant/handler tools: 85+ tests covering dispatch, registry, resolvers, routes, llm_planner, 13 tool handlers - clean release: artifact_catalog_loader, mappers, approval, publication tests - API routes: translate_helpers, validation_service extensions, datasets to 100% - notifications: providers/service tests - services: profile_preference_service - docs/orthogonal-test-report.md — full speckit.tests audit - Fixes: 3 git_base async mock failures, 4 assistant handler permission-check patches - .gitignore: coverage artifacts
414 lines
16 KiB
Python
414 lines
16 KiB
Python
# #region Test.CleanRelease.Repositories [C:3] [TYPE Module] [SEMANTICS test,clean-release,repository,persistence,sqlalchemy]
|
|
# @BRIEF Tests for all clean release SQLAlchemy repositories via mocked sessions.
|
|
# @RELATION BINDS_TO -> [clean_release_repositories]
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
|
|
from src.models.clean_release import (
|
|
ApprovalDecision,
|
|
CandidateArtifact,
|
|
CleanPolicySnapshot,
|
|
CleanReleaseAuditLog,
|
|
ComplianceDecision,
|
|
ComplianceReport,
|
|
ComplianceRun,
|
|
ComplianceStageRun,
|
|
ComplianceViolation,
|
|
DistributionManifest,
|
|
PublicationRecord,
|
|
ReleaseCandidate,
|
|
RunStatus,
|
|
SourceRegistrySnapshot,
|
|
)
|
|
from src.services.clean_release.enums import CandidateStatus
|
|
from src.services.clean_release.repositories import (
|
|
ApprovalRepository,
|
|
ArtifactRepository,
|
|
AuditRepository,
|
|
CandidateRepository,
|
|
ComplianceRepository,
|
|
ManifestRepository,
|
|
PolicyRepository,
|
|
PublicationRepository,
|
|
ReportRepository,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db():
|
|
"""Create a mock SQLAlchemy Session."""
|
|
return MagicMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_query(mock_db):
|
|
"""Configure mock_db.query to return a mock query chain."""
|
|
query_mock = MagicMock()
|
|
mock_db.query.return_value = query_mock
|
|
return query_mock
|
|
|
|
|
|
# === ApprovalRepository ===
|
|
|
|
class TestApprovalRepository:
|
|
"""ApprovalRepository — save, get_by_id, get_latest, list."""
|
|
|
|
# #region test_save [C:2] [TYPE Function]
|
|
def test_save(self, mock_db):
|
|
repo = ApprovalRepository(mock_db)
|
|
decision = ApprovalDecision(id="a1", candidate_id="c1", report_id="r1", decision="APPROVED", decided_by="tester", decided_at=datetime.now(UTC))
|
|
result = repo.save(decision)
|
|
mock_db.add.assert_called_with(decision)
|
|
mock_db.commit.assert_called_once()
|
|
mock_db.refresh.assert_called_with(decision)
|
|
assert result == decision
|
|
# #endregion test_save
|
|
|
|
# #region test_get_by_id [C:2] [TYPE Function]
|
|
def test_get_by_id(self, mock_db, mock_query):
|
|
repo = ApprovalRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
result = repo.get_by_id("a1")
|
|
assert result is not None
|
|
mock_query.filter.assert_called_once()
|
|
# #endregion test_get_by_id
|
|
|
|
# #region test_get_latest_for_candidate [C:2] [TYPE Function]
|
|
def test_get_latest_for_candidate(self, mock_db, mock_query):
|
|
repo = ApprovalRepository(mock_db)
|
|
mock_filter = mock_query.filter.return_value
|
|
mock_filter.order_by.return_value.first.return_value = MagicMock()
|
|
result = repo.get_latest_for_candidate("c1")
|
|
assert result is not None
|
|
mock_filter.order_by.assert_called_once()
|
|
# #endregion test_get_latest_for_candidate
|
|
|
|
# #region test_list_by_candidate [C:2] [TYPE Function]
|
|
def test_list_by_candidate(self, mock_db, mock_query):
|
|
repo = ApprovalRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
result = repo.list_by_candidate("c1")
|
|
assert len(result) == 1
|
|
# #endregion test_list_by_candidate
|
|
|
|
|
|
# === ArtifactRepository ===
|
|
|
|
class TestArtifactRepository:
|
|
"""ArtifactRepository — save, save_all, get_by_id, list_by_candidate."""
|
|
|
|
# #region test_save [C:2] [TYPE Function]
|
|
def test_save(self, mock_db):
|
|
repo = ArtifactRepository(mock_db)
|
|
art = CandidateArtifact(id="a1", candidate_id="c1", path="p", sha256="s", size=1)
|
|
result = repo.save(art)
|
|
mock_db.add.assert_called_with(art)
|
|
mock_db.commit.assert_called_once()
|
|
assert result == art
|
|
# #endregion test_save
|
|
|
|
# #region test_save_all [C:2] [TYPE Function]
|
|
def test_save_all(self, mock_db):
|
|
repo = ArtifactRepository(mock_db)
|
|
arts = [MagicMock(), MagicMock()]
|
|
result = repo.save_all(arts)
|
|
mock_db.add_all.assert_called_with(arts)
|
|
mock_db.commit.assert_called_once()
|
|
assert result == arts
|
|
# #endregion test_save_all
|
|
|
|
# #region test_get_by_id [C:2] [TYPE Function]
|
|
def test_get_by_id(self, mock_db, mock_query):
|
|
repo = ArtifactRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_by_id("a1") is not None
|
|
# #endregion test_get_by_id
|
|
|
|
# #region test_list_by_candidate [C:2] [TYPE Function]
|
|
def test_list_by_candidate(self, mock_db, mock_query):
|
|
repo = ArtifactRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
assert len(repo.list_by_candidate("c1")) == 1
|
|
# #endregion test_list_by_candidate
|
|
|
|
|
|
# === AuditRepository ===
|
|
|
|
class TestAuditRepository:
|
|
"""AuditRepository — log, list_by_candidate."""
|
|
|
|
# #region test_log [C:2] [TYPE Function]
|
|
def test_log(self, mock_db):
|
|
repo = AuditRepository(mock_db)
|
|
mock_db.add.return_value = None
|
|
result = repo.log(action="PREP", actor="tester", candidate_id="c1", details={"k": "v"})
|
|
mock_db.add.assert_called_once()
|
|
mock_db.commit.assert_called_once()
|
|
assert isinstance(result, CleanReleaseAuditLog)
|
|
# #endregion test_log
|
|
|
|
# #region test_log_no_details [C:2] [TYPE Function]
|
|
def test_log_no_details(self, mock_db):
|
|
repo = AuditRepository(mock_db)
|
|
result = repo.log(action="CHECK", actor="system")
|
|
assert result.details_json == {}
|
|
# #endregion test_log_no_details
|
|
|
|
# #region test_list_by_candidate [C:2] [TYPE Function]
|
|
def test_list_by_candidate(self, mock_db, mock_query):
|
|
repo = AuditRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = []
|
|
assert repo.list_by_candidate("c1") == []
|
|
# #endregion test_list_by_candidate
|
|
|
|
|
|
# === CandidateRepository ===
|
|
|
|
class TestCandidateRepository:
|
|
"""CandidateRepository — save, get_by_id, list_all."""
|
|
|
|
# #region test_save [C:2] [TYPE Function]
|
|
def test_save(self, mock_db):
|
|
repo = CandidateRepository(mock_db)
|
|
cand = ReleaseCandidate(id="c1", version="1", source_snapshot_ref="r", created_by="t", created_at=datetime.now(UTC), status=CandidateStatus.DRAFT.value)
|
|
result = repo.save(cand)
|
|
mock_db.add.assert_called_with(cand)
|
|
mock_db.commit.assert_called_once()
|
|
assert result == cand
|
|
# #endregion test_save
|
|
|
|
# #region test_get_by_id [C:2] [TYPE Function]
|
|
def test_get_by_id(self, mock_db, mock_query):
|
|
repo = CandidateRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_by_id("c1") is not None
|
|
# #endregion test_get_by_id
|
|
|
|
# #region test_list_all [C:2] [TYPE Function]
|
|
def test_list_all(self, mock_db, mock_query):
|
|
repo = CandidateRepository(mock_db)
|
|
mock_query.all.return_value = [MagicMock()]
|
|
assert len(repo.list_all()) == 1
|
|
# #endregion test_list_all
|
|
|
|
|
|
# === ComplianceRepository ===
|
|
|
|
class TestComplianceRepository:
|
|
"""ComplianceRepository — run, stage, violation CRUD."""
|
|
|
|
# #region test_save_run [C:2] [TYPE Function]
|
|
def test_save_run(self, mock_db):
|
|
repo = ComplianceRepository(mock_db)
|
|
run = ComplianceRun(id="r1", candidate_id="c1", manifest_id="m1", manifest_digest="d1", policy_snapshot_id="p1", registry_snapshot_id="r1", requested_by="tester", requested_at=datetime.now(UTC), started_at=datetime.now(UTC), status=RunStatus.RUNNING.value)
|
|
result = repo.save_run(run)
|
|
mock_db.add.assert_called_with(run)
|
|
mock_db.commit.assert_called_once()
|
|
assert result == run
|
|
# #endregion test_save_run
|
|
|
|
# #region test_get_run [C:2] [TYPE Function]
|
|
def test_get_run(self, mock_db, mock_query):
|
|
repo = ComplianceRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_run("r1") is not None
|
|
# #endregion test_get_run
|
|
|
|
# #region test_list_runs_by_candidate [C:2] [TYPE Function]
|
|
def test_list_runs_by_candidate(self, mock_db, mock_query):
|
|
repo = ComplianceRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
assert len(repo.list_runs_by_candidate("c1")) == 1
|
|
# #endregion test_list_runs_by_candidate
|
|
|
|
# #region test_save_stage_run [C:2] [TYPE Function]
|
|
def test_save_stage_run(self, mock_db):
|
|
repo = ComplianceRepository(mock_db)
|
|
sr = ComplianceStageRun(id="s1", run_id="r1", stage_name="DP", status="SUCCEEDED", decision="PASSED", details_json={})
|
|
result = repo.save_stage_run(sr)
|
|
mock_db.add.assert_called_with(sr)
|
|
assert result == sr
|
|
# #endregion test_save_stage_run
|
|
|
|
# #region test_list_stages_by_run [C:2] [TYPE Function]
|
|
def test_list_stages_by_run(self, mock_db, mock_query):
|
|
repo = ComplianceRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
assert len(repo.list_stages_by_run("r1")) == 1
|
|
# #endregion test_list_stages_by_run
|
|
|
|
# #region test_save_violation [C:2] [TYPE Function]
|
|
def test_save_violation(self, mock_db):
|
|
repo = ComplianceRepository(mock_db)
|
|
v = ComplianceViolation(id="v1", run_id="r1", stage_name="DP", code="C", severity="MAJOR", message="m")
|
|
result = repo.save_violation(v)
|
|
mock_db.add.assert_called_with(v)
|
|
assert result == v
|
|
# #endregion test_save_violation
|
|
|
|
# #region test_save_violations_batch [C:2] [TYPE Function]
|
|
def test_save_violations_batch(self, mock_db):
|
|
repo = ComplianceRepository(mock_db)
|
|
vs = [MagicMock(), MagicMock()]
|
|
result = repo.save_violations(vs)
|
|
mock_db.add_all.assert_called_with(vs)
|
|
assert result == vs
|
|
# #endregion test_save_violations_batch
|
|
|
|
# #region test_list_violations_by_run [C:2] [TYPE Function]
|
|
def test_list_violations_by_run(self, mock_db, mock_query):
|
|
repo = ComplianceRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
assert len(repo.list_violations_by_run("r1")) == 1
|
|
# #endregion test_list_violations_by_run
|
|
|
|
|
|
# === ManifestRepository ===
|
|
|
|
class TestManifestRepository:
|
|
"""ManifestRepository — save, get_by_id, get_latest, list."""
|
|
|
|
# #region test_save [C:2] [TYPE Function]
|
|
def test_save(self, mock_db):
|
|
repo = ManifestRepository(mock_db)
|
|
m = DistributionManifest(id="m1", candidate_id="c1", manifest_version=1, manifest_digest="d1", artifacts_digest="d1", source_snapshot_ref="r", content_json={}, created_by="t", created_at=datetime.now(UTC))
|
|
result = repo.save(m)
|
|
mock_db.add.assert_called_with(m)
|
|
assert result == m
|
|
# #endregion test_save
|
|
|
|
# #region test_get_by_id [C:2] [TYPE Function]
|
|
def test_get_by_id(self, mock_db, mock_query):
|
|
repo = ManifestRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_by_id("m1") is not None
|
|
# #endregion test_get_by_id
|
|
|
|
# #region test_get_latest_for_candidate [C:2] [TYPE Function]
|
|
def test_get_latest_for_candidate(self, mock_db, mock_query):
|
|
repo = ManifestRepository(mock_db)
|
|
mock_filter = mock_query.filter.return_value
|
|
mock_filter.order_by.return_value.first.return_value = MagicMock()
|
|
assert repo.get_latest_for_candidate("c1") is not None
|
|
# #endregion test_get_latest_for_candidate
|
|
|
|
# #region test_list_by_candidate [C:2] [TYPE Function]
|
|
def test_list_by_candidate(self, mock_db, mock_query):
|
|
repo = ManifestRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
assert len(repo.list_by_candidate("c1")) == 1
|
|
# #endregion test_list_by_candidate
|
|
|
|
|
|
# === PolicyRepository ===
|
|
|
|
class TestPolicyRepository:
|
|
"""PolicyRepository — policy/registry snapshot CRUD."""
|
|
|
|
# #region test_save_policy [C:2] [TYPE Function]
|
|
def test_save_policy(self, mock_db):
|
|
repo = PolicyRepository(mock_db)
|
|
p = CleanPolicySnapshot(id="p1", policy_id="p1", policy_version="1", content_json={}, registry_snapshot_id="r1", immutable=True)
|
|
result = repo.save_policy_snapshot(p)
|
|
mock_db.add.assert_called_with(p)
|
|
assert result == p
|
|
# #endregion test_save_policy
|
|
|
|
# #region test_get_policy [C:2] [TYPE Function]
|
|
def test_get_policy(self, mock_db, mock_query):
|
|
repo = PolicyRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_policy_snapshot("p1") is not None
|
|
# #endregion test_get_policy
|
|
|
|
# #region test_save_registry [C:2] [TYPE Function]
|
|
def test_save_registry(self, mock_db):
|
|
repo = PolicyRepository(mock_db)
|
|
r = SourceRegistrySnapshot(id="r1", registry_id="r1", registry_version="1", allowed_hosts=["h"], immutable=True)
|
|
result = repo.save_registry_snapshot(r)
|
|
mock_db.add.assert_called_with(r)
|
|
assert result == r
|
|
# #endregion test_save_registry
|
|
|
|
# #region test_get_registry [C:2] [TYPE Function]
|
|
def test_get_registry(self, mock_db, mock_query):
|
|
repo = PolicyRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_registry_snapshot("r1") is not None
|
|
# #endregion test_get_registry
|
|
|
|
|
|
# === PublicationRepository ===
|
|
|
|
class TestPublicationRepository:
|
|
"""PublicationRepository — save, get_by_id, get_latest, list."""
|
|
|
|
# #region test_save [C:2] [TYPE Function]
|
|
def test_save(self, mock_db):
|
|
repo = PublicationRepository(mock_db)
|
|
p = PublicationRecord(id="p1", candidate_id="c1", report_id="r1", published_by="t", published_at=datetime.now(UTC), target_channel="s", status="ACTIVE")
|
|
result = repo.save(p)
|
|
mock_db.add.assert_called_with(p)
|
|
assert result == p
|
|
# #endregion test_save
|
|
|
|
# #region test_get_by_id [C:2] [TYPE Function]
|
|
def test_get_by_id(self, mock_db, mock_query):
|
|
repo = PublicationRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_by_id("p1") is not None
|
|
# #endregion test_get_by_id
|
|
|
|
# #region test_get_latest [C:2] [TYPE Function]
|
|
def test_get_latest(self, mock_db, mock_query):
|
|
repo = PublicationRepository(mock_db)
|
|
mock_filter = mock_query.filter.return_value
|
|
mock_filter.order_by.return_value.first.return_value = MagicMock()
|
|
assert repo.get_latest_for_candidate("c1") is not None
|
|
# #endregion test_get_latest
|
|
|
|
# #region test_list [C:2] [TYPE Function]
|
|
def test_list(self, mock_db, mock_query):
|
|
repo = PublicationRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
assert len(repo.list_by_candidate("c1")) == 1
|
|
# #endregion test_list
|
|
|
|
|
|
# === ReportRepository ===
|
|
|
|
class TestReportRepository:
|
|
"""ReportRepository — save, get_by_id, get_by_run, list."""
|
|
|
|
# #region test_save [C:2] [TYPE Function]
|
|
def test_save(self, mock_db):
|
|
repo = ReportRepository(mock_db)
|
|
r = ComplianceReport(id="CCR-1", run_id="run-1", candidate_id="c1", generated_at=datetime.now(UTC), final_status=ComplianceDecision.PASSED.value, summary_json={}, immutable=True)
|
|
result = repo.save(r)
|
|
mock_db.add.assert_called_with(r)
|
|
assert result == r
|
|
# #endregion test_save
|
|
|
|
# #region test_get_by_id [C:2] [TYPE Function]
|
|
def test_get_by_id(self, mock_db, mock_query):
|
|
repo = ReportRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_by_id("CCR-1") is not None
|
|
# #endregion test_get_by_id
|
|
|
|
# #region test_get_by_run [C:2] [TYPE Function]
|
|
def test_get_by_run(self, mock_db, mock_query):
|
|
repo = ReportRepository(mock_db)
|
|
mock_query.filter.return_value.first.return_value = MagicMock()
|
|
assert repo.get_by_run("run-1") is not None
|
|
# #endregion test_get_by_run
|
|
|
|
# #region test_list_by_candidate [C:2] [TYPE Function]
|
|
def test_list_by_candidate(self, mock_db, mock_query):
|
|
repo = ReportRepository(mock_db)
|
|
mock_query.filter.return_value.all.return_value = [MagicMock()]
|
|
assert len(repo.list_by_candidate("c1")) == 1
|
|
# #endregion test_list_by_candidate
|