garden_layer/test_garden_layer.py

import os
import pytest
import requests
from typing import Any, Dict

from garden_layer import GardenLayer
from garden_layer.config import CLONE_OF, GRAPH, KEROSENE_ROOT, MCP_URL

CYCLE_2026_3 = "http://world.eu.org/example1#ProductionCycle_21d96b6c-179"


def pytest_report_header(config):
    return "Garden layer tests: requires virtuoso_mcp running at GardenLayer.MCP_URL"


@pytest.fixture
def garden_layer():
    layer = GardenLayer()
    try:
        layer.traverse(KEROSENE_ROOT, CLONE_OF, direction="incoming", limit=1)
    except requests.RequestException as exc:
        pytest.skip(f"Cannot reach MCP server: {exc}")
    return layer


def test_traverse_clone_tree(garden_layer):
    """Walk `cloneOf` incoming edges and ensure the root has clones to inspect."""
    result = garden_layer.traverse(KEROSENE_ROOT, CLONE_OF, direction="incoming", limit=5)
    assert isinstance(result, dict)
    bindings = result.get("results", {}).get("bindings", [])
    assert isinstance(bindings, list)
    assert bindings, "Expected at least one clone binding"
    print(f"Traverse of {KEROSENE_ROOT} returned {len(bindings)} clone candidates.")


def test_describe_subject(garden_layer):
    """Describe the root plant and surface the top predicates the garden helpers can reuse."""
    summary = garden_layer.describe_subject(KEROSENE_ROOT, limit=5)
    assert isinstance(summary, dict)
    bindings = summary.get("results", {}).get("bindings", [])
    assert bindings, "Expected describe_subject to return bindings"
    triples = [f"{b['predicate']['value']} -> {b['object'].get('value', '<literal>')}" for b in bindings]
    print("describe_subject returned:")
    for triple in triples:
        print("  ", triple)


def test_path_traverse_lineage(garden_layer):
    """Use the path helper to follow `cloneOf` and observe the lineage step."""
    path = garden_layer.path_traverse(KEROSENE_ROOT, [CLONE_OF], direction="incoming", limit=5)
    assert isinstance(path, dict)
    bindings = path.get("result", {}).get("results", {}).get("bindings", [])
    assert isinstance(bindings, list)
    assert bindings, "Path traverse should find at least one step"
    print(f"Property path {CLONE_OF} produced {len(bindings)} step bindings.")


def test_property_usage_statistics(garden_layer):
    """Summarize how the `cloneOf` property is used so future helpers can rely on frequency data."""
    stats = garden_layer.property_usage_statistics(CLONE_OF, examples_limit=3)
    assert isinstance(stats, dict)
    count_bindings = stats.get("count", {}).get("results", {}).get("bindings", [])
    assert count_bindings, "Expected usage count bindings"
    usage_count = count_bindings[0].get("usageCount", {}).get("value")
    example_bindings = stats.get("examples", {}).get("results", {}).get("bindings", [])
    assert isinstance(example_bindings, list)
    print(f"cloneOf usage count: {usage_count}")
    print("example bindings:")
    for binding in example_bindings:
        subject = binding.get("subjectLabel", {}).get("value") or binding.get("subject", {}).get("value")
        object_ = binding.get("objectLabel", {}).get("value") or binding.get("object", {}).get("value")
        print(f"  - {subject} -> {object_}")


def test_batch_insert(garden_layer):
    """Batch-insert a TTL snippet and verify the query shape"""
    ttl = '<http://world.eu.org/example1#batch_test_subject> <http://www.w3.org/2000/01/rdf-schema#label> "garden batch" .'
    result = garden_layer.batch_insert(ttl=ttl, graph=GRAPH)
    assert isinstance(result, dict)
    assert "query" in result
    assert "INSERT DATA" in result.get("query", "").upper()
    print("batch_insert generated:")
    print(result.get("query", ""))


def call_mcp_tool(tool_name: str, payload: Dict[str, Any]) -> Any:
    response = requests.post(MCP_URL, json={"tool": tool_name, "input": payload}, timeout=10)
    response.raise_for_status()
    body = response.json()
    assert body.get("status") == "ok", f"{tool_name} failed: {body.get('detail') or 'unknown reason'}"
    return body["result"]


def test_mcp_garden_describe_subject():
    result = call_mcp_tool(
        "garden_describe_subject",
        {"subject_uri": KEROSENE_ROOT, "limit": 5},
    )
    bindings = result.get("results", {}).get("bindings", [])
    assert bindings, "garden_describe_subject should return bindings"
    print("garden_describe_subject returned", len(bindings), "bindings via /mcp")


def test_mcp_garden_property_usage_statistics():
    result = call_mcp_tool(
        "garden_property_usage_statistics",
        {"property_uri": CLONE_OF, "examples_limit": 2},
    )
    count_bindings = result.get("count", {}).get("results", {}).get("bindings", [])
    assert count_bindings, "Expected usage count bindings from the garden tool"
    examples = result.get("examples", {}).get("results", {}).get("bindings", [])
    assert isinstance(examples, list)
    print("garden_property_usage_statistics sample bindings:", len(examples))


def test_mcp_garden_cycle_plants():
    result = call_mcp_tool(
        "garden_cycle_plants",
        {"cycle_uri": CYCLE_2026_3, "limit": 5},
    )
    bindings = result.get("results", {}).get("bindings", [])
    assert bindings, "Expected at least one plant binding from the cycle"
    print("garden_cycle_plants returned", len(bindings), "plants for cycle", CYCLE_2026_3)
    for binding in bindings:
        plant_uri = binding.get("plant", {}).get("value")
        assert plant_uri, "Each binding should include a plant URI"


def test_load_examples_fixture(garden_layer):
    if os.getenv("MCP_ALLOW_EXAMPLE_LOAD", "false").lower() != "true":
        pytest.skip("MCP_ALLOW_EXAMPLE_LOAD must be true to load example fixtures")
    result = garden_layer.load_examples(files=["test_fixture.ttl"], graph=GRAPH)
    loaded = result.get("loaded", [])
    assert any(item.get("file") == "test_fixture.ttl" for item in loaded), "Expected fixture file to be loaded"