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