garden_layer/test_garden_layer.py

import pytest
import requests

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


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", ""))