lore-engine-poc/tests/test_multi_world.py

"""
Tests for v2.T6 — multi-world namespace (world_id on all nodes).

These tests verify the multi-world semantics declared in
lore-engine/docs/01-ontology.md:
  - Every node carries a `world_id` property.
  - Every tool that touches nodes filters by `world_id`.
  - `list_worlds()` returns the distinct world ids in the graph.
  - Two parallel worlds (default, arda_greyscale) coexist without
    cross-contamination.
"""
import os
import sys

import pytest

ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
for p in (os.path.join(ROOT, "gateway"), os.path.join(ROOT, "plugins")):
    if p not in sys.path:
        sys.path.insert(0, p)


NEO4J_URL = os.environ.get("TEST_NEO4J_URL", "bolt://localhost:7687")
NEO4J_USER = os.environ.get("TEST_NEO4J_USER", "neo4j")
NEO4J_PASSWORD = os.environ.get("TEST_NEO4J_PASSWORD", "lore-dev-password")
PG_URL = os.environ.get(
    "TEST_PG_URL", "postgresql://lore:***@localhost:5433/lore"
)


def _ensure_neo4j_env():
    os.environ.setdefault("NEO4J_URL", NEO4J_URL)
    os.environ.setdefault("NEO4J_USER", NEO4J_USER)
    os.environ.setdefault("NEO4J_PASSWORD", NEO4J_PASSWORD)
    os.environ.setdefault("POSTGRES_URL", PG_URL)
    os.environ.setdefault("MINIO_URL", "http://localhost:9000")
    os.environ.setdefault("MINIO_ACCESS_KEY", "lorelore")
    os.environ.setdefault("MINIO_SECRET_KEY", "lore-dev-password")
    os.environ.setdefault("MINIO_BUCKET", "lore-images")
    os.environ.setdefault("MINIO_PUBLIC_URL", "http://localhost:9000")


def _neo4j_session():
    from neo4j import GraphDatabase
    return GraphDatabase.driver(NEO4J_URL, auth=(NEO4J_USER, NEO4J_PASSWORD))


@pytest.fixture
def neo4j():
    _ensure_neo4j_env()
    drv = _neo4j_session()
    with drv.session() as s:
        s.run("MATCH (n:TestWorld) DETACH DELETE n")
    yield drv
    with drv.session() as s:
        s.run("MATCH (n:TestWorld) DETACH DELETE n")
    drv.close()


# ─── list_worlds ─────────────────────────────────────────────────────────────

def test_list_worlds_returns_distinct_world_ids(neo4j):
    """list_worlds() must return the distinct world_id values present in
    the graph. After seed.py runs, the graph has at least 'default' and
    'arda_greyscale'."""
    from plugins.world import list_worlds
    rows = list_worlds({})
    world_ids = {r["world_id"] for r in rows if r.get("world_id")}
    # Both worlds must be present after seeding.
    assert "default" in world_ids, f"missing 'default' in {world_ids}"
    assert "arda_greyscale" in world_ids, f"missing 'arda_greyscale' in {world_ids}"


# ─── entity_context world_id filter ──────────────────────────────────────────

def test_entity_context_world_id_default(neo4j):
    """entity_context in 'default' world returns the default-world Aldric."""
    from plugins.world import entity_context
    res = entity_context({"name": "Aldric Raventhorne", "world_id": "default"})
    assert res.get("found") is True
    assert res["id"] == "aldric"


def test_entity_context_world_id_isolation(neo4j):
    """entity_context in 'arda_greyscale' for a name that only exists in
    'default' must return {found: false}."""
    from plugins.world import entity_context
    res = entity_context({"name": "Aldric Raventhorne", "world_id": "arda_greyscale"})
    assert res.get("found") is False, f"leaked across worlds: {res}"


def test_entity_context_world_id_arda_greyscale(neo4j):
    """entity_context in 'arda_greyscale' for a name that exists only
    there returns that entity."""
    from plugins.world import entity_context
    res = entity_context({"name": "Mael Greyscale", "world_id": "arda_greyscale"})
    assert res.get("found") is True, f"not found in greyscale world: {res}"
    assert res["id"] == "mael_greyscale"


def test_entity_context_default_param_is_default_world(neo4j):
    """Calling entity_context without world_id must default to 'default'."""
    from plugins.world import entity_context
    res = entity_context({"name": "Aldric Raventhorne"})
    assert res.get("found") is True
    assert res["id"] == "aldric"


# ─── was_true_at world_id filter ─────────────────────────────────────────────

def test_was_true_at_world_id_filter(neo4j):
    """A relation that exists in 'default' must not match in 'arda_greyscale'."""
    from plugins.world import was_true_at
    res_default = was_true_at({
        "relation": "ALLIED_WITH",
        "subject": "House Vyr",
        "object": "Merchants Guild",
        "at_time": "2nd_age.year_230",
        "world_id": "default",
    })
    assert res_default.get("was_true") is True, f"default world should be true: {res_default}"
    res_gs = was_true_at({
        "relation": "ALLIED_WITH",
        "subject": "House Vyr",
        "object": "Merchants Guild",
        "at_time": "2nd_age.year_230",
        "world_id": "arda_greyscale",
    })
    assert res_gs.get("was_true") is False, f"greyscale world should be false: {res_gs}"


# ─── state_at world_id filter ────────────────────────────────────────────────

def test_state_at_world_id_filter(neo4j):
    """state_at in 'arda_greyscale' for a name only in 'default' must not find it."""
    from plugins.world import state_at
    res = state_at({
        "entity": "Aldric Raventhorne",
        "at_time": "2nd_age.year_260",
        "world_id": "arda_greyscale",
    })
    assert res.get("found") is False, f"leaked across worlds: {res}"


# ─── ancestors_of / descendants_of world_id filter ──────────────────────────

def test_ancestors_of_world_id_filter(neo4j):
    """ancestors_of in 'arda_greyscale' for a default-world person must return empty."""
    from plugins.lineage import ancestors_of
    res = ancestors_of({"person": "Aldric Raventhorne", "generations": 5, "world_id": "arda_greyscale"})
    assert res["ancestors"] == [], f"leaked: {res}"


def test_descendants_of_world_id_filter(neo4j):
    """descendants_of in 'arda_greyscale' for a default-world person must return empty."""
    from plugins.lineage import descendants_of
    res = descendants_of({"person": "Theron Ashveil", "generations": 5, "world_id": "arda_greyscale"})
    assert res["descendants"] == [], f"leaked: {res}"


def test_lineage_of_world_id_filter(neo4j):
    """lineage_of in 'arda_greyscale' for a default-world person must return found=false."""
    from plugins.lineage import lineage_of
    res = lineage_of({"person": "Aldric Raventhorne", "world_id": "arda_greyscale"})
    assert res.get("found") is False, f"leaked: {res}"


# ─── recall_images / search_images_by_caption world_id filter ───────────────

def test_recall_images_world_id_filter(neo4j):
    """recall_images for an arda_greyscale person must return that world's images only."""
    from plugins.images import recall_images
    res = recall_images({"entity_id": "mael_greyscale", "world_id": "arda_greyscale"})
    assert res["count"] >= 1, f"no images for greyscale person: {res}"
    for img in res["images"]:
        assert img.get("world_id") == "arda_greyscale", f"wrong world: {img}"


def test_recall_images_world_id_isolation(neo4j):
    """recall_images for a default-world person in 'arda_greyscale' must return 0."""
    from plugins.images import recall_images
    res = recall_images({"entity_id": "aldric", "world_id": "arda_greyscale"})
    assert res["count"] == 0, f"leaked: {res}"


def test_search_images_by_caption_world_id_filter(neo4j):
    """search_images_by_caption for 'greyscale' in 'default' must return 0."""
    from plugins.images import search_images_by_caption
    res = search_images_by_caption({"q": "greyscale", "world_id": "default"})
    assert res["count"] == 0, f"leaked: {res}"
    res_gs = search_images_by_caption({"q": "greyscale", "world_id": "arda_greyscale"})
    assert res_gs["count"] >= 1, f"missing greyscale matches: {res_gs}"


# ─── world_id column on image_manifest ───────────────────────────────────────

def test_image_manifest_has_world_id_column(neo4j):
    """image_manifest must have a world_id column populated for every row."""
    import psycopg2
    conn = psycopg2.connect(PG_URL)
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT column_name FROM information_schema.columns
                WHERE table_name = 'image_manifest' AND column_name = 'world_id'
            """)
            assert cur.fetchone() is not None, "image_manifest is missing world_id column"
    finally:
        conn.close()