148 lines
4.4 KiB
Python
148 lines
4.4 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
from ont_platform.core.review import (
|
|
CandidatePromotionService,
|
|
EvidenceRequiredError,
|
|
InvalidReviewTransitionError,
|
|
ReviewService,
|
|
)
|
|
from ont_platform.storage.candidate_repository import CandidateRepository
|
|
from ont_platform.storage.models import Base, CandidateEntity, CandidateKind, ReviewStatus
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
|
|
def _session() -> Session:
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(engine)
|
|
return sessionmaker(bind=engine)()
|
|
|
|
|
|
def _seed_candidate(
|
|
repository: CandidateRepository,
|
|
*,
|
|
candidate_id: str = "E_review",
|
|
evidence_id: str = "EV_review",
|
|
with_evidence: bool = True,
|
|
) -> CandidateEntity:
|
|
payload = {
|
|
"entities": [
|
|
{
|
|
"id": candidate_id,
|
|
"label": "Review",
|
|
"type": "concept",
|
|
"confidence": 0.92,
|
|
"source_trust": 0.9,
|
|
"validation_passed": True,
|
|
"evidence_ids": [evidence_id] if with_evidence else [],
|
|
}
|
|
],
|
|
"relations": [],
|
|
"evidence_spans": [
|
|
{
|
|
"id": evidence_id,
|
|
"text": "Review candidates need evidence.",
|
|
"start_offset": 0,
|
|
"end_offset": 32,
|
|
}
|
|
]
|
|
if with_evidence
|
|
else [],
|
|
}
|
|
return repository.save_lightweight_result(
|
|
project_id="proj_1",
|
|
document_id="doc_1",
|
|
result=payload,
|
|
source_trust=0.9,
|
|
validation_passed=True,
|
|
).entities[0]
|
|
|
|
|
|
def test_approve_requires_valid_evidence_and_records_decision() -> None:
|
|
db = _session()
|
|
repository = CandidateRepository(db)
|
|
entity = _seed_candidate(repository)
|
|
|
|
decision = ReviewService(repository).approve(
|
|
candidate_kind=CandidateKind.ENTITY,
|
|
candidate_id=entity.id,
|
|
reviewed_by="lasta",
|
|
reason="looks good",
|
|
)
|
|
|
|
assert entity.review_status == ReviewStatus.APPROVED
|
|
assert decision.previous_status == ReviewStatus.PENDING
|
|
assert decision.new_status == ReviewStatus.APPROVED
|
|
assert len(repository.review_history(candidate_kind=CandidateKind.ENTITY, candidate_id=entity.id)) == 1
|
|
|
|
|
|
def test_candidate_without_evidence_cannot_be_approved() -> None:
|
|
db = _session()
|
|
repository = CandidateRepository(db)
|
|
entity = _seed_candidate(repository, with_evidence=False)
|
|
|
|
with pytest.raises(EvidenceRequiredError):
|
|
ReviewService(repository).approve(
|
|
candidate_kind=CandidateKind.ENTITY,
|
|
candidate_id=entity.id,
|
|
reviewed_by="lasta",
|
|
)
|
|
|
|
assert entity.review_status == ReviewStatus.PENDING
|
|
|
|
|
|
def test_auto_approve_requires_policy_thresholds() -> None:
|
|
db = _session()
|
|
repository = CandidateRepository(db)
|
|
entity = _seed_candidate(repository)
|
|
|
|
decision = ReviewService(repository).auto_approve(
|
|
candidate_kind=CandidateKind.ENTITY,
|
|
candidate_id=entity.id,
|
|
)
|
|
|
|
assert decision.new_status == ReviewStatus.AUTO_APPROVED
|
|
assert entity.review_status == ReviewStatus.AUTO_APPROVED
|
|
|
|
|
|
def test_rejected_candidate_cannot_be_approved_again() -> None:
|
|
db = _session()
|
|
repository = CandidateRepository(db)
|
|
entity = _seed_candidate(repository)
|
|
service = ReviewService(repository)
|
|
|
|
service.reject(
|
|
candidate_kind=CandidateKind.ENTITY,
|
|
candidate_id=entity.id,
|
|
reviewed_by="lasta",
|
|
)
|
|
|
|
with pytest.raises(InvalidReviewTransitionError):
|
|
service.approve(
|
|
candidate_kind=CandidateKind.ENTITY,
|
|
candidate_id=entity.id,
|
|
reviewed_by="lasta",
|
|
)
|
|
|
|
|
|
def test_promotion_plan_blocks_approved_candidate_without_evidence() -> None:
|
|
db = _session()
|
|
repository = CandidateRepository(db)
|
|
valid = _seed_candidate(repository)
|
|
invalid = _seed_candidate(repository, candidate_id="E_no_evidence", with_evidence=False)
|
|
invalid.review_status = ReviewStatus.APPROVED
|
|
valid.review_status = ReviewStatus.APPROVED
|
|
|
|
plan = CandidatePromotionService(repository).build_commit_plan(project_id="proj_1")
|
|
|
|
assert [entity.id for entity in plan.entities] == ["E_review"]
|
|
assert plan.blocked == [
|
|
{
|
|
"candidate_kind": "entity",
|
|
"candidate_id": "E_no_evidence",
|
|
"reason": "missing_or_invalid_evidence",
|
|
"review_status": "approved",
|
|
}
|
|
]
|