mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
233 lines
9.3 KiB
Python
233 lines
9.3 KiB
Python
"""Regression tests for the standalone mcp/ package export_graph tool.
|
|
|
|
The mcp/ server (python -m mcp / python -m mcp.server) had two failures on
|
|
every RDF export format:
|
|
|
|
1. AttributeError: 'ContextGraph' object has no attribute 'get'
|
|
handle_export_graph() in mcp/tools/export.py called
|
|
RDFExporter().export_to_rdf(graph, ...) passing the raw ContextGraph
|
|
object instead of the canonical kg dict expected by the exporter.
|
|
|
|
2. stdout progress corruption
|
|
RDFExporter.__init__ instantiated the Semantica progress-tracker
|
|
singleton, which wrote a progress bar to sys.stdout before the
|
|
AttributeError was raised. stdout is the MCP stdio JSON-RPC transport,
|
|
so this interleaved non-JSON bytes corrupted framing for every client.
|
|
|
|
Fixes applied:
|
|
- mcp/tools/export.py: convert with graph.to_kg_dict() before export_to_rdf()
|
|
- mcp/__init__.py: os.environ["SEMANTICA_DISABLE_PROGRESS"] = "1" at
|
|
package initialisation, before any tool handler can instantiate
|
|
RDFExporter and therefore before the tracker singleton is created.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import os
|
|
import sys
|
|
import subprocess
|
|
import unittest
|
|
|
|
import semantica.utils.progress_tracker as _progress_module
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_graph():
|
|
"""Return a ContextGraph with two entities and one relationship."""
|
|
from semantica.context.context_graph import ContextGraph
|
|
g = ContextGraph()
|
|
g.add_node("n1", node_type="entity")
|
|
g.add_node("n2", node_type="entity")
|
|
g.add_edge("n1", "n2", "related_to")
|
|
return g
|
|
|
|
|
|
def _reset_progress_singleton():
|
|
"""Destroy any cached progress-tracker singleton so the next call
|
|
to get_progress_tracker() reads the current environment variable."""
|
|
_progress_module.ProgressTracker._instance = None
|
|
_progress_module._global_tracker = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RDF export correctness
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMCPPackageExportGraphRDF(unittest.TestCase):
|
|
"""handle_export_graph() must return a non-empty RDF string for every
|
|
supported RDF format, not an error dict."""
|
|
|
|
def setUp(self):
|
|
# Inject a known graph into the mcp/ session so handlers don't try to
|
|
# build a full ContextGraph (which requires heavy ML dependencies).
|
|
import mcp.session as _session
|
|
self._orig_graph = _session._graph
|
|
_session._graph = _make_graph()
|
|
|
|
def tearDown(self):
|
|
import mcp.session as _session
|
|
_session._graph = self._orig_graph
|
|
|
|
def test_turtle_returns_non_empty_string(self):
|
|
from mcp.tools.export import handle_export_graph
|
|
result = handle_export_graph({"format": "turtle"})
|
|
self.assertNotIn("error", result, result)
|
|
self.assertIsInstance(result["data"], str)
|
|
self.assertGreater(len(result["data"]), 0)
|
|
# Turtle output must carry prefix declarations
|
|
self.assertIn("@prefix", result["data"])
|
|
|
|
def test_ttl_alias_returns_non_empty_string(self):
|
|
from mcp.tools.export import handle_export_graph
|
|
result = handle_export_graph({"format": "ttl"})
|
|
self.assertNotIn("error", result, result)
|
|
self.assertIsInstance(result["data"], str)
|
|
self.assertGreater(len(result["data"]), 0)
|
|
|
|
def test_nt_returns_non_empty_string(self):
|
|
from mcp.tools.export import handle_export_graph
|
|
result = handle_export_graph({"format": "nt"})
|
|
self.assertNotIn("error", result, result)
|
|
self.assertIsInstance(result["data"], str)
|
|
self.assertGreater(len(result["data"]), 0)
|
|
|
|
def test_xml_returns_non_empty_string(self):
|
|
from mcp.tools.export import handle_export_graph
|
|
result = handle_export_graph({"format": "xml"})
|
|
self.assertNotIn("error", result, result)
|
|
self.assertIsInstance(result["data"], str)
|
|
self.assertGreater(len(result["data"]), 0)
|
|
|
|
def test_jsonld_returns_non_empty_string(self):
|
|
from mcp.tools.export import handle_export_graph
|
|
result = handle_export_graph({"format": "json-ld"})
|
|
self.assertNotIn("error", result, result)
|
|
self.assertIsInstance(result["data"], str)
|
|
self.assertGreater(len(result["data"]), 0)
|
|
|
|
def test_all_rdf_formats_succeed(self):
|
|
from mcp.tools.export import handle_export_graph
|
|
for fmt in ("turtle", "ttl", "nt", "xml", "json-ld"):
|
|
with self.subTest(fmt=fmt):
|
|
result = handle_export_graph({"format": fmt})
|
|
self.assertNotIn("error", result, f"format={fmt}: {result}")
|
|
self.assertIsInstance(result["data"], str)
|
|
self.assertGreater(len(result["data"]), 0)
|
|
|
|
def test_rdf_branch_does_not_raise_context_graph_attribute_error(self):
|
|
"""The pre-fix code passed ContextGraph directly to export_to_rdf(),
|
|
causing AttributeError: 'ContextGraph' object has no attribute 'get'.
|
|
Verify that error does not appear in the result."""
|
|
from mcp.tools.export import handle_export_graph
|
|
result = handle_export_graph({"format": "turtle"})
|
|
if "error" in result:
|
|
self.assertNotIn("'ContextGraph' object has no attribute 'get'",
|
|
result["error"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# stdout protection — subprocess-based to avoid process-state cross-contamination
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMCPPackageStdoutProtection(unittest.TestCase):
|
|
"""The standalone mcp/ server must not write any progress bytes to stdout.
|
|
stdout is the MCP JSON-RPC transport channel.
|
|
|
|
These tests use a subprocess to get a clean process state where
|
|
SEMANTICA_DISABLE_PROGRESS has not yet been set, so we can verify that
|
|
importing mcp and running an export produces no progress bytes on stdout.
|
|
"""
|
|
|
|
def _run_in_subprocess(self, code: str, timeout: int = 30) -> subprocess.CompletedProcess:
|
|
"""Run a Python snippet in a clean subprocess with the repo on sys.path."""
|
|
repo_root = os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..")
|
|
)
|
|
env = os.environ.copy()
|
|
env["PYTHONPATH"] = repo_root
|
|
# Start with a clean slate — no pre-set disable flag
|
|
env.pop("SEMANTICA_DISABLE_PROGRESS", None)
|
|
return subprocess.run(
|
|
[sys.executable, "-c", code],
|
|
cwd=repo_root,
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
timeout=timeout,
|
|
check=False,
|
|
)
|
|
|
|
def test_importing_mcp_sets_disable_progress(self):
|
|
"""Importing the mcp package must set SEMANTICA_DISABLE_PROGRESS=1
|
|
before any tool handler runs."""
|
|
code = (
|
|
"import os; "
|
|
"import mcp; " # triggers mcp/__init__.py
|
|
"print(os.environ.get('SEMANTICA_DISABLE_PROGRESS', 'NOT SET'))"
|
|
)
|
|
result = self._run_in_subprocess(code)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertIn("1", result.stdout)
|
|
|
|
def test_rdf_export_writes_no_progress_to_stdout(self):
|
|
"""An RDF export via handle_export_graph() must not write any Semantica
|
|
progress bytes to stdout. The only stdout bytes should be the explicit
|
|
print() call at the end of the snippet."""
|
|
code = """
|
|
import os, sys
|
|
# Ensure clean state
|
|
os.environ.pop("SEMANTICA_DISABLE_PROGRESS", None)
|
|
|
|
import mcp # sets SEMANTICA_DISABLE_PROGRESS=1
|
|
import mcp.session as session
|
|
from semantica.context.context_graph import ContextGraph
|
|
|
|
g = ContextGraph()
|
|
g.add_node("n1", node_type="entity")
|
|
g.add_node("n2", node_type="entity")
|
|
g.add_edge("n1", "n2", "related_to")
|
|
session._graph = g
|
|
|
|
# Intercept stdout writes to detect any progress output
|
|
written = []
|
|
_orig = sys.stdout.write
|
|
def _capture(s):
|
|
written.append(s)
|
|
return _orig(s)
|
|
sys.stdout.write = _capture
|
|
|
|
from mcp.tools.export import handle_export_graph
|
|
result = handle_export_graph({"format": "turtle"})
|
|
|
|
sys.stdout.write = _orig
|
|
|
|
# Only our explicit output below should be in written
|
|
# (the sentinel line is added after restoring stdout)
|
|
progress_writes = [s for s in written]
|
|
print("RESULT_OK:" + str("error" not in result))
|
|
print("STDOUT_WRITES:" + str(len(progress_writes)))
|
|
"""
|
|
proc = self._run_in_subprocess(code)
|
|
self.assertEqual(proc.returncode, 0, proc.stderr)
|
|
# Extract the printed lines
|
|
lines = proc.stdout.strip().splitlines()
|
|
result_ok_line = next((l for l in lines if l.startswith("RESULT_OK:")), None)
|
|
writes_line = next((l for l in lines if l.startswith("STDOUT_WRITES:")), None)
|
|
self.assertIsNotNone(result_ok_line, f"stdout: {proc.stdout!r}")
|
|
self.assertIsNotNone(writes_line, f"stdout: {proc.stdout!r}")
|
|
self.assertEqual(result_ok_line, "RESULT_OK:True",
|
|
f"export returned error; stdout={proc.stdout!r}, stderr={proc.stderr!r}")
|
|
n_writes = int(writes_line.split(":")[1])
|
|
self.assertEqual(n_writes, 0,
|
|
f"Expected 0 progress writes to stdout, got {n_writes}; "
|
|
f"stdout={proc.stdout!r}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|