tests/test_claim_validator.py

from __future__ import annotations

from copy import deepcopy

from claim_validation.schemas import ValidateClaimsRequest
from claim_validation.service import validate_claims
from extraction.schemas import AbstractArticle, ExtractionContext
from extraction.service import extract_claim_candidates


ABSTRACT = (
    "BACKGROUND: Rosemary has been traditionally used for memory. "
    "METHODS: Sixty-eight university students were randomized to oral rosemary or placebo. "
    "RESULTS: Oral rosemary 500 mg twice a day for 1 month significantly improved prospective and retrospective memory compared with placebo (p < 0.05). "
    "CONCLUSIONS: Rosemary may improve selected memory outcomes in university students."
)


def candidate() -> dict:
    article = AbstractArticle.model_validate({
        "pmid": "29389474",
        "title": "Effects of Rosmarinus officinalis L. on memory performance in university students",
        "abstract": ABSTRACT,
        "year": 2018,
        "journal": "Complementary Therapies in Clinical Practice",
        "publication_types": ["Randomized Controlled Trial"],
        "mesh_terms": ["Rosmarinus", "Memory"],
        "language": "en",
    })
    result = extract_claim_candidates(
        article,
        ExtractionContext(focus_substance="rosemary", focus_scientific_name="Rosmarinus officinalis"),
    )
    return deepcopy(result["claims"][0])


def request_for(claim: dict, *, abstract: str = ABSTRACT, language: str = "en", translations=None):
    return ValidateClaimsRequest.model_validate({
        "claims": [claim],
        "source_abstracts": [{"pmid": "29389474", "abstract": abstract, "language": language}],
        "translations": translations or [],
    })


def supported_runner(claim, sources, entailment_threshold, contradiction_threshold):
    return {
        "verdict": "SUPPORTED",
        "reason": "supported in exact source quote",
        "entailment": 0.93,
        "neutral": 0.05,
        "contradiction": 0.02,
    }


def partial_runner(claim, sources, entailment_threshold, contradiction_threshold):
    return {
        "verdict": "PARTIAL",
        "reason": "partial support",
        "entailment": 0.58,
        "neutral": 0.35,
        "contradiction": 0.07,
    }


def contradicted_runner(claim, sources, entailment_threshold, contradiction_threshold):
    return {
        "verdict": "CONTRADICTED",
        "reason": "source contradicts claim",
        "entailment": 0.05,
        "neutral": 0.10,
        "contradiction": 0.85,
    }


def test_supported_numeric_claim_becomes_validated() -> None:
    result = validate_claims(request_for(candidate()), nli_runner=supported_runner)
    claim = result["claims"][0]
    assert result["validated_count"] == 1
    assert claim["claim_status"] == "validated"
    assert claim["validation"]["provenance_status"] == "verified"
    assert claim["validation"]["numeric_status"] == "validated"
    assert claim["validation"]["entailment_status"] == "entailed"
    assert all(fact["status"] == "validated" for fact in claim["numeric_facts"])
    assert all(fact["correction"] is None for fact in claim["numeric_facts"])


def test_provenance_mismatch_is_rejected() -> None:
    result = validate_claims(
        request_for(candidate(), abstract="This is a different abstract without the exact result sentence."),
        nli_runner=supported_runner,
    )
    assert result["rejected_count"] == 1
    assert result["claims"][0]["claim_status"] == "rejected"
    assert result["claims"][0]["validation"]["provenance_status"] == "rejected"


def test_numeric_mismatch_is_conflicted_without_correction() -> None:
    claim = candidate()
    claim["claim_text"] = claim["claim_text"].replace("500 mg", "50 mg")
    for fact in claim["numeric_facts"]:
        if fact["kind"] == "dose":
            fact["raw_text"] = "50 mg"
            fact["value"] = 50.0
    result = validate_claims(request_for(claim), nli_runner=supported_runner)
    validated = result["claims"][0]
    assert result["conflicted_count"] == 1
    assert validated["claim_status"] == "conflicted"
    assert validated["validation"]["numeric_status"] == "conflicted"
    assert any(fact["status"] == "conflicted" for fact in validated["numeric_facts"])
    assert all(fact["correction"] is None for fact in validated["numeric_facts"])


def test_partial_support_remains_visible_but_not_validated() -> None:
    result = validate_claims(request_for(candidate()), nli_runner=partial_runner)
    assert result["partial_count"] == 1
    assert result["claims"][0]["claim_status"] == "partial"
    assert result["claims"][0]["confidence"]["overall"] == "low"


def test_contradiction_blocks_claim() -> None:
    result = validate_claims(request_for(candidate()), nli_runner=contradicted_runner)
    assert result["conflicted_count"] == 1
    claim = result["claims"][0]
    assert claim["validation"]["contradiction_status"] == "contradicted"
    assert claim["validation"]["contradicting_pmids"] == ["29389474"]


def test_non_english_claim_requires_verified_translation() -> None:
    claim = candidate()
    claim["source"]["source_language"] = "fr"
    claim["validation"]["translation_status"] = "translated_unverified"
    result = validate_claims(request_for(claim, language="fr"), nli_runner=supported_runner)
    assert result["rejected_count"] == 1
    assert result["claims"][0]["validation"]["translation_status"] == "missing_verified_translation"


def test_verified_translation_allows_nli() -> None:
    claim = candidate()
    claim["source"]["source_language"] = "fr"
    claim["validation"]["translation_status"] = "translated_unverified"
    translations = [{
        "claim_id": claim["claim_id"],
        "claim_en": claim["claim_text"],
        "evidence_en": claim["source"]["source_quote"],
        "status": "verified",
        "engine": "test-translator",
    }]
    result = validate_claims(
        request_for(claim, language="fr", translations=translations),
        nli_runner=supported_runner,
    )
    assert result["validated_count"] == 1
    assert result["claims"][0]["validation"]["translation_status"] == "verified"


def test_high_confidence_is_not_assigned_before_replication() -> None:
    result = validate_claims(request_for(candidate()), nli_runner=supported_runner)
    assert result["claims"][0]["confidence"]["overall"] != "high"
    assert result["claims"][0]["confidence"]["replication"] == "not_applicable"