waxwing/tests/test_knowledge_ingest.py

"""cf-harvest knowledge ingest tests — critical path.

Covers: all 8 fact types, idempotent re-ingest, partial batch failure,
subject extraction, and fact_hash deduplication.
"""
import pytest


_SOURCE = {
    "video_path": "/media/GardenersWorld/S59E01.mkv",
    "presenter": "Monty Don",
    "location": "Longmeadow",
    "source_domain": "gardening",
}

_ALL_FACT_TYPES = [
    ("companion_planting", {
        "plants": ["carrot", "spring onion"],
        "relationship": "beneficial",
        "benefit": "deters carrot root fly",
        "spacing": "alternate rows",
        "notes": None,
    }),
    ("soil_amendment", {
        "amendment": "pine bark mulch",
        "purpose": "suppress weeds, retain moisture",
        "application_rate": "5cm layer",
        "timing": "early spring",
        "target_plants": ["all beds"],
    }),
    ("propagation", {
        "plant": "raspberry",
        "method": "cane",
        "timing": "autumn",
        "steps": ["cut cane to 30cm", "push into prepared soil", "firm in"],
        "success_indicators": ["new leaves at base"],
        "notes": None,
    }),
    ("pest_diagnosis", {
        "pest_or_disease": "couch grass",
        "affected_plants": ["border perennials"],
        "symptoms": ["white rhizomes through roots", "vigorous re-sprouting"],
        "treatment": "fork out every rhizome fragment",
        "prevention": "mulch to reduce light",
        "organic": True,
    }),
    ("harvest_timing", {
        "plant": "raspberry",
        "indicators": ["deep red colour", "comes free with gentle pull"],
        "harvest_method": "gentle pull",
        "post_harvest": "refrigerate within 24h",
    }),
    ("instruction", {
        "action": "mulch",
        "subject": "border",
        "steps": ["clear weeds", "apply 5cm pine bark"],
        "timing": "early spring",
        "notes": None,
    }),
    ("medicinal", {
        "plant": "Calendula officinalis",
        "use": "wound healing, anti-inflammatory skin salve",
        "preparation": "infused oil",
        "cautions": "avoid during pregnancy",
    }),
    ("history_context", {
        "subject": "Narcissus triandrus",
        "origin": "Iberian Peninsula",
        "period": None,
        "summary": "Angel's Tears — a small, nodding daffodil native to rocky hillsides.",
    }),
]


def test_ingest_all_eight_fact_types(client):
    facts = [{"fact_type": ft, "confidence": 0.9, "payload": p} for ft, p in _ALL_FACT_TYPES]
    resp = client.post("/api/v1/knowledge/glean", json={"source": _SOURCE, "facts": facts})
    assert resp.status_code == 200
    data = resp.json()
    assert data["facts_received"] == 8
    assert data["facts_inserted"] == 8
    assert data["facts_skipped_duplicate"] == 0
    assert data["rejected"] == []


def test_ingest_idempotent(client):
    facts = [{"fact_type": "instruction", "confidence": 0.8, "payload": {
        "action": "water", "subject": "tomatoes", "steps": [], "timing": "morning",
    }}]
    payload = {"source": _SOURCE, "facts": facts}

    first = client.post("/api/v1/knowledge/glean", json=payload).json()
    assert first["facts_inserted"] == 1

    second = client.post("/api/v1/knowledge/glean", json=payload).json()
    assert second["facts_inserted"] == 0
    assert second["facts_skipped_duplicate"] == 1


def test_ingest_partial_batch_continues_on_bad_fact(client):
    facts = [
        {"fact_type": "instruction", "confidence": 0.9, "payload": {
            "action": "prune", "subject": "roses", "steps": [], "timing": "spring",
        }},
        {"fact_type": "unknown_type", "confidence": 0.5, "payload": {}},
        {"fact_type": "medicinal", "confidence": 0.7, "payload": {
            "plant": "Lavender", "use": "calming", "preparation": "tea", "cautions": None,
        }},
    ]
    resp = client.post("/api/v1/knowledge/glean", json={"source": _SOURCE, "facts": facts})
    assert resp.status_code == 422


def test_ingest_valid_and_empty_payload(client):
    facts = [{"fact_type": "history_context", "confidence": 0.6, "payload": {
        "subject": "Wisteria", "origin": "China", "period": "Tang Dynasty", "summary": "Introduced to Europe in 1816.",
    }}]
    resp = client.post("/api/v1/knowledge/glean", json={"source": _SOURCE, "facts": facts})
    assert resp.status_code == 200
    assert resp.json()["facts_inserted"] == 1


def test_ingest_source_is_upserted_across_calls(client):
    facts = [{"fact_type": "instruction", "confidence": 0.8, "payload": {
        "action": "dig", "subject": "bed", "steps": [], "timing": "autumn",
    }}]
    r1 = client.post("/api/v1/knowledge/glean", json={"source": _SOURCE, "facts": facts}).json()
    r2 = client.post("/api/v1/knowledge/glean", json={"source": _SOURCE, "facts": facts}).json()
    assert r1["source_id"] == r2["source_id"]


def test_ingest_subject_extracted_for_companion_planting(client):
    facts = [{"fact_type": "companion_planting", "confidence": 0.95, "payload": {
        "plants": ["basil", "tomato"], "relationship": "beneficial",
        "benefit": "repels aphids", "spacing": None, "notes": None,
    }}]
    resp = client.post("/api/v1/knowledge/glean", json={"source": _SOURCE, "facts": facts})
    assert resp.status_code == 200

    query = client.get("/api/v1/knowledge?fact_type=companion_planting")
    assert query.status_code == 200
    facts_out = query.json()
    assert len(facts_out) == 1
    assert facts_out[0]["subject"] == "basil"