lore-engine-poc/tests/test_consistency.py

"""
Tests for the consistency plugin (v2.T5).

These tests exercise the 4 tools directly via the plugin module (not via the
HTTP gateway), talking to the same Neo4j the gateway uses. The seed data
(5 hand-crafted violations) is the contract — counts must match:

    find_contradictions()        -> count = 1
    find_anachronisms()          -> count = 1
    find_orphans()               -> count = 1
    find_ontology_violations()   -> count = 2
    total                        = 5

The tools MUST run real detection (not just return empty envelopes) — the
test asserts that each tool surfaces a violation whose `id` matches a seeded
one, and that severity/status/details fields are populated.
"""
import os
import sys
import pytest

# Make gateway/ + plugins/ importable (matches the pattern in other tests).
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
# Put our local paths FIRST so the local `plugins/` and `gateway/` packages
# win over any venv-installed `plugins` package.
for p in (os.path.join(ROOT, "plugins"), os.path.join(ROOT, "gateway")):
    if p not in sys.path:
        sys.path.insert(0, p)

# Connection env defaults — explicitly OVERWRITE so the shell's redacted
# `NEO4J_PASSWORD=***` placeholder doesn't win over the real password.
os.environ["NEO4J_URL"]     = os.environ.get("NEO4J_URL", "bolt://localhost:7687")
os.environ["NEO4J_USER"]    = os.environ.get("NEO4J_USER", "neo4j")
if not os.environ.get("NEO4J_PASSWORD") or len(os.environ["NEO4J_PASSWORD"]) < 8:
    os.environ["NEO4J_PASSWORD"] = "lore-dev-password"

from plugins import consistency  # noqa: E402


def _shape_ok(v):
    """A violation dict has id/severity/status/details (and maybe more)."""
    assert isinstance(v, dict), f"violation is not a dict: {v!r}"
    for key in ("id", "severity", "status", "details"):
        assert key in v, f"violation missing {key!r}: {v!r}"


# ─── find_contradictions ─────────────────────────────────────────────────────

def test_find_contradictions_returns_one():
    """The seeded contradiction (Aldric's overlapping memberships) is found."""
    res = consistency.find_contradictions({})
    assert res["count"] == 1, f"expected 1 contradiction, got {res!r}"
    assert len(res["violations"]) == 1
    _shape_ok(res["violations"][0])
    assert res["violations"][0]["id"] == "c_aldric_double_membership"


def test_find_contradictions_severity_filter():
    """severity='error' returns only the error-severity contradiction;
    severity='warn' returns none; severity='any' (default) returns 1."""
    any_res = consistency.find_contradictions({"severity": "any"})
    err_res = consistency.find_contradictions({"severity": "error"})
    warn_res = consistency.find_contradictions({"severity": "warn"})
    assert any_res["count"] == 1
    assert err_res["count"] == 1
    assert warn_res["count"] == 0


# ─── find_anachronisms ───────────────────────────────────────────────────────

def test_find_anachronisms_returns_one():
    """Vex (born 180) at the Founding of House Vyr (year 85) is the seeded
    anachronism. The tool must surface it."""
    res = consistency.find_anachronisms({})
    assert res["count"] == 1, f"expected 1 anachronism, got {res!r}"
    _shape_ok(res["violations"][0])
    assert res["violations"][0]["id"] == "a_vex_at_founding"
    # The details string should mention the year math so the LLM/operator
    # can see *why* this is an anachronism without re-querying.
    assert "180" in res["violations"][0]["details"]
    assert "85" in res["violations"][0]["details"]


# ─── find_orphans ────────────────────────────────────────────────────────────

def test_find_orphans_returns_only_lyssa():
    """The hand-crafted orphan is Lyssa the Watcher. Other People / Items
    have at least one relation (the v2.T5 fix-up rows in seed.py ensure this)."""
    res = consistency.find_orphans({})
    assert res["count"] == 1, f"expected 1 orphan, got {res!r}"
    _shape_ok(res["violations"][0])
    assert res["violations"][0]["id"] == "o_unfinished_npc"
    # The orphan must point at lyssa, not some other entity.
    assert "lyssa" in res["violations"][0]["details"].lower() or \
           "watcher" in res["violations"][0]["details"].lower()


# ─── find_ontology_violations ────────────────────────────────────────────────

def test_find_ontology_violations_returns_two():
    """theron and maric are missing death years despite being born < 280.
    The rule 'persons_born_before_280_must_die' fires on both → 2 violations."""
    res = consistency.find_ontology_violations({})
    assert res["count"] == 2, f"expected 2 ontology violations, got {res!r}"
    ids = {v["id"] for v in res["violations"]}
    assert "ov_theron_no_died" in ids
    assert "ov_maric_no_died" in ids
    for v in res["violations"]:
        _shape_ok(v)
        # The seeded rule id appears in the violation so callers can link
        # back to the OntologyRule that triggered the finding.
        assert "persons_born_before_280_must_die" in v.get("rule_id", ""), \
            f"violation missing rule_id: {v!r}"


def test_ontology_rule_node_exists():
    """The OntologyRule node must exist in the graph for the tool to consume."""
    from neo4j import GraphDatabase
    d = GraphDatabase.driver(os.environ["NEO4J_URL"],
                             auth=(os.environ["NEO4J_USER"],
                                   os.environ["NEO4J_PASSWORD"]))
    try:
        with d.session() as s:
            row = s.run("""
                MATCH (r:OntologyRule {id: 'persons_born_before_280_must_die'})
                RETURN r.id AS id, r.severity AS severity, r.cutoff_year AS cutoff
            """).single()
        assert row is not None, "OntologyRule node missing"
        assert row["severity"] == "warn"
        assert row["cutoff"] == 280
    finally:
        d.close()


# ─── envelope shape (all 4 tools) ────────────────────────────────────────────

@pytest.mark.parametrize("tool_name,args", [
    ("find_contradictions",      {}),
    ("find_anachronisms",        {}),
    ("find_orphans",             {}),
    ("find_ontology_violations", {}),
])
def test_tool_envelope_shape(tool_name, args):
    """Every tool returns the {violations: [...], count: N} envelope."""
    fn = getattr(consistency, tool_name)
    res = fn(args)
    assert isinstance(res, dict)
    assert "violations" in res and isinstance(res["violations"], list)
    assert "count" in res and isinstance(res["count"], int)
    assert res["count"] == len(res["violations"]), \
        f"count != len(violations): {res!r}"