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"
FIXTURE_FILES = ["ganja_test_fixture.ttl"]


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}")
    if os.getenv("MCP_ALLOW_EXAMPLE_LOAD", "false").lower() == "true":
        try:
            layer.load_examples(files=FIXTURE_FILES, graph=GRAPH)
        except Exception as exc:
            pytest.skip(f"Cannot load garden fixtures: {exc}")
    result = layer.traverse(KEROSENE_ROOT, CLONE_OF, direction="incoming", limit=1)
    bindings = result.get("results", {}).get("bindings", [])
    if not bindings:
        pytest.skip("No garden fixture data available; set MCP_ALLOW_EXAMPLE_LOAD=true")
    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:
    # Some ontology/traversal calls can be a bit slow depending on Virtuoso load.
    response = requests.post(MCP_URL, json={"tool": tool_name, "input": payload}, timeout=20)
    if response.status_code >= 400:
        raise AssertionError(
            f"{tool_name} failed with {response.status_code}: {response.text}"
        )
    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": 1},
    )
    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": 1},
    )
    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_mcp_garden_cycle_plants_latest_when_cycle_uri_missing():
    """MCP compliance: cycle_uri should be optional and default to latest cycle."""
    result = call_mcp_tool(
        "garden_cycle_plants",
        {"limit": 3},
    )
    bindings = result.get("results", {}).get("bindings", [])
    assert bindings, "Expected plant bindings from the latest cycle when cycle_uri is omitted"


def test_mcp_garden_cycle_list_detailed_has_strain_info():
    """Ensure the detailed cycle list includes strain labels/descriptions and clone siblings."""
    result = call_mcp_tool(
        "garden_cycle_list_detailed",
        {"limit": 10},
    )
    bindings = result.get("results", {}).get("bindings", [])
    assert bindings, "Expected at least one binding from garden_cycle_list_detailed"

    # Check that at least one binding has strainDesc or cloneSiblings populated.
    any_strain_desc = any('strainDesc' in b and b.get('strainDesc', {}).get('value') for b in bindings)
    any_siblings = any('cloneSiblings' in b and b.get('cloneSiblings', {}).get('value') for b in bindings)
    assert any_strain_desc or any_siblings, "Expected strainDesc and/or cloneSiblings to be present"


def test_mcp_input_schema_announced_for_garden_tools():
    """OpenClaw MCP compliance: garden_* tools should expose inputSchema via tools/list."""
    response = requests.post(
        MCP_URL,
        json={"jsonrpc": "2.0", "id": 999, "method": "tools/list", "params": {}},
        timeout=10,
    )
    response.raise_for_status()
    body = response.json()
    assert body.get("result"), f"Expected tools/list result, got: {body}"

    tools = body["result"].get("tools") or []
    tool_by_name = {t.get("name"): t for t in tools if isinstance(t, dict)}

    # At least validate a couple of the “new” tools.
    for name in [
        "garden_clone_to",
        "garden_cycle_plants",
        "garden_cycle_list_detailed",
    ]:
        assert name in tool_by_name, f"Missing tool {name} in tools/list"
        input_schema = tool_by_name[name].get("inputSchema")
        assert isinstance(input_schema, dict), f"{name}.inputSchema missing/invalid"

    # Specifically: garden_cycle_plants should allow missing cycle_uri.
    cycle_plants_schema = tool_by_name["garden_cycle_plants"].get("inputSchema", {})
    props = cycle_plants_schema.get("properties", {})
    assert "cycle_uri" in props, "garden_cycle_plants inputSchema must define cycle_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"