tests/test_substitution_v11.py

from __future__ import annotations

from pino.substitution_v11 import evaluate_formula, evaluate_material, evaluate_retriever, score_candidate, substitute, wilson_interval


def tiny_profiles():
    return [
        {
            "pimt_version": "v11",
            "cas": "1",
            "name": "Target Floral",
            "descriptor_distribution": {"F": 1.0},
            "substantivity_log10_predicted": 1.0,
            "log10_vapor_pressure_pa": 0.5,
            "boiling_point_k": 450.0,
            "vapor_pressure_pa": 3.0,
            "enrichment_pending": ["IFRA/allergen/use-level"],
        },
        {
            "pimt_version": "v11",
            "cas": "2",
            "name": "Documented Floral Substitute",
            "descriptor_distribution": {"F": 0.9, "G": 0.1},
            "substantivity_log10_predicted": 1.05,
            "log10_vapor_pressure_pa": 0.4,
            "boiling_point_k": 455.0,
            "vapor_pressure_pa": 2.5,
            "enrichment_pending": ["IFRA/allergen/use-level"],
        },
        {
            "pimt_version": "v11",
            "cas": "3",
            "name": "Wrong Citrus",
            "descriptor_distribution": {"A": 1.0},
            "substantivity_log10_predicted": 3.0,
            "log10_vapor_pressure_pa": 4.0,
            "boiling_point_k": 330.0,
            "vapor_pressure_pa": 1000.0,
            "enrichment_pending": ["IFRA/allergen/use-level"],
        },
    ]


def test_substitute_ranks_documented_like_profile_first():
    ranked = substitute("1", tiny_profiles(), mode="function_match", top_k=2)

    assert ranked[0]["candidate_cas"] == "2"
    assert ranked[0]["score"] > ranked[1]["score"]


def test_evaluate_retriever_uses_eval_rows_without_training():
    rows = [
        {
            "pair_id": "pair-1",
            "target": "Target Floral",
            "target_cas": "1",
            "substitute": "Documented Floral Substitute",
            "substitute_cas": "2",
            "grade": "acceptable",
            "mode": "odor_match",
        }
    ]

    result = evaluate_retriever(rows, tiny_profiles(), top_k=1)

    assert result["n_evaluable"] == 1
    assert result["n_hit"] == 1
    assert result["n_missed"] == 0
    assert result["n_excluded"] == 0
    assert result["top_k_hit"] == wilson_interval(1, 1)
    assert result["mode"] == "per-row"
    assert result["mode_counts"] == {"odor_match": 1}
    assert result["by_mode_top_k_hit"] == {"odor_match": wilson_interval(1, 1)}
    assert "trained" not in result


def test_evaluate_retriever_uses_each_rows_mode():
    rows = [
        {
            "pair_id": "odor-row",
            "target_cas": "1",
            "substitute_cas": "2",
            "grade": "acceptable",
            "mode": "odor_match",
        },
        {
            "pair_id": "function-row",
            "target_cas": "1",
            "substitute_cas": "2",
            "grade": "acceptable",
            "mode": "function_match",
        },
    ]

    result = evaluate_retriever(rows, tiny_profiles(), top_k=1)

    assert result["mode_counts"] == {"function_match": 1, "odor_match": 1}
    assert [item["mode"] for item in result["evaluated"]] == ["odor_match", "function_match"]


def test_accord_requires_every_component_and_reports_component_recall():
    rows = [
        {
            "pair_id": "accord-row",
            "target_cas": "1",
            "grade": "acceptable",
            "mode": "accord_rebuild",
            "answer": {
                "shape": "accord",
                "materials": [{"cas": "2"}, {"cas": "3"}],
            },
        }
    ]

    missed = evaluate_retriever(rows, tiny_profiles(), top_k=1)
    hit = evaluate_retriever(rows, tiny_profiles(), top_k=2)

    assert missed["n_missed"] == 1
    assert missed["evaluated"][0]["answer_components_retrieved"] == 1
    assert missed["evaluated"][0]["answer_components_total"] == 2
    assert hit["n_hit"] == 1


def test_any_of_accepts_one_preregistered_alternative():
    rows = [
        {
            "pair_id": "any-of-row",
            "target_cas": "1",
            "grade": "acceptable",
            "mode": "function_match",
            "answer": {
                "shape": "any_of",
                "materials": [{"cas": "2"}, {"cas": "3"}],
            },
        }
    ]

    result = evaluate_retriever(rows, tiny_profiles(), top_k=1)

    assert result["n_hit"] == 1
    assert result["evaluated"][0]["answer_components_retrieved"] == 1


def test_excluded_and_missed_are_separate_first_class_counts():
    rows = [
        {
            "pair_id": "excluded",
            "target_cas": "missing-target",
            "substitute_cas": "2",
            "grade": "acceptable",
            "mode": "function_match",
        },
        {
            "pair_id": "missed",
            "target_cas": "1",
            "substitute_cas": "3",
            "grade": "acceptable",
            "mode": "function_match",
        },
    ]

    result = evaluate_retriever(rows, tiny_profiles(), top_k=1)

    assert result["n_eval_rows"] == 2
    assert result["n_excluded"] == 1
    assert result["n_evaluable"] == 1
    assert result["n_missed"] == 1
    assert result["n_hit"] == 0


def test_assessment_interfaces_do_not_emit_safety_authority():
    material = evaluate_material("1", tiny_profiles())
    formula = evaluate_formula([{"cas": "1", "weight_fraction": 0.4}], tiny_profiles())

    assert material["pimt_version"] == "v11"
    assert "IFRA/allergen/use-level" in material["enrichment_pending"]
    assert formula["substitution_risks"] == ["safety/use-level flags enrichment-pending; no safety claims emitted"]


def test_missing_features_are_unobserved_not_neutral_positive_evidence():
    scored = score_candidate(tiny_profiles()[0], {"cas": "4", "name": "Empty"})

    assert scored["score"] == 0.0
    assert scored["evidence_coverage"] == 0.0
    assert all(value is None for value in scored["reasons"].values())


def test_answer_resolving_to_target_is_invalid_not_a_miss():
    rows = [{
        "pair_id": "self-answer", "target_cas": "1", "substitute_cas": "1",
        "grade": "acceptable", "mode": "function_match",
    }]

    result = evaluate_retriever(rows, tiny_profiles())

    assert result["n_evaluable"] == 0
    assert result["excluded"] == [{"pair_id": "self-answer", "reason": "answer_resolves_to_target"}]


def test_unimplemented_mode_is_reported_unsupported_not_as_a_miss():
    rows = [{
        "pair_id": "cost-without-cost-data", "target_cas": "1", "substitute_cas": "2",
        "grade": "acceptable", "mode": "cost_match",
    }]

    result = evaluate_retriever(rows, tiny_profiles())

    assert result["n_evaluable"] == 0
    assert result["excluded"][0]["reason"] == "mode_features_unsupported"


def test_cost_mode_uses_observed_cost_axis_when_available():
    target = tiny_profiles()[0] | {"cost_tier": "low"}
    same = tiny_profiles()[1] | {"cost_tier": "low"}
    different = tiny_profiles()[2] | {"cost_tier": "high"}

    assert score_candidate(target, same, mode="cost_match")["reasons"]["constraints"] == 1.0
    assert score_candidate(target, different, mode="cost_match")["reasons"]["constraints"] == 0.0