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', '')}" 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 = ' "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"