AI/ontology_platform/tests/unit/test_review_service.py

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",
        }
    ]