mengzhx/semantica
1
0
Fork 0
mirror of https://github.com/semantica-agi/semantica.git synced 2026-09-06 04:00:19 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
dc306079ea837ebed4c69ef6bbf4ecc1e3e4676c
semantica/tests/explorer/test_explorer_api.py
T
13g4d013g4d0Sameer Kadam
 3254b9be80 Serve the RDF export formats the MCP tool already offers (#1131) (#1157) 
* feat(explorer): serve the RDF export formats the MCP tool already offers (#1131)

`POST /api/export` accepted only `json` and `csv` and answered 422 for everything
else, while the MCP `export_graph` tool resolved Turtle, N-Triples, RDF/XML,
JSON-LD, GraphML and Parquet through `semantica.export`. Two surfaces of one
product disagreeing about what the product can do — and for an RDF-native project,
a graph that loads as JSON-LD and cannot be exported as RDF is a one-way door.

The route now reaches the same exporters the MCP tool uses. Nothing is
reimplemented: `RDFExporter.export_to_rdf` and `GraphMLExporter.export` receive the
dict `session.build_graph_dict()` already builds.

The alias table is a copy of `mcp/tools/export.py::_FORMAT_ALIASES` plus the
spellings the issue mentioned (`ntriples`, `rdf-xml`), and a test asserts the two
tables agree — if either drifts, the formats a caller can use would depend on which
door they came through.

Media types and extensions per serializer, so a Turtle export is `text/turtle` and
not `application/json` with a `.json` name.

The 422 message now names what IS supported. The old one said only that the format
was unsupported, which reads as "this format does not exist" rather than "this door
does not open it" — that is what sent me looking through the library.

Parquet is left out on purpose: `ParquetExporter.export` writes a file and returns a
path, so serving it over HTTP is a different shape of change and deserves its own
review.

Tests, in `TestImportExport`: the seven RDF spellings, each **parsed with rdflib**
rather than asserted on strings — a response that merely looks like Turtle is what
lets this class of gap survive a suite. Plus the alias-agreement canary and the
error message. Three mutations (rejecting RDF again, breaking one alias, emptying
the message) each turn the matching tests red.

110 tests in `tests/explorer/test_explorer_api.py` pass.

* fix(explorer): complete RDF export support

* fix(explorer): secure GraphML temporary file handling

* fix(ci): scan declared dependencies with Safety

---------

Co-authored-by: 13g4d0 <13g4d0@users.noreply.github.com>
Co-authored-by: Sameer Kadam <sskadam6305@gmail.com>
2026-09-01 15:01:13 +05:30

1476 lines
58 KiB
Python
Raw Blame History

"""Integration tests for the explorer API."""
from datetime import datetime
import json
from pathlib import Path
import uuid
import networkx as nx
import pytest
from semantica.context.context_graph import ContextGraph
# fastapi ships in the optional `explorer` extra, not in `dev`, so this module
# must skip rather than fail collection when it is absent. The guard has to sit
# above the import below, which pulls fastapi in transitively.
pytest.importorskip("fastapi")
from semantica.explorer.app import create_app # noqa: E402
from semantica.explorer.session import GraphSession # noqa: E402
from starlette.testclient import TestClient # noqa: E402
def _build_sample_graph() -> ContextGraph:
graph = ContextGraph(advanced_analytics=False)
graph.add_node(
"python",
node_type="language",
content="Python programming language",
popularity="high",
x=10,
y=15,
tags=["lang", "featured"],
)
graph.add_node("javascript", node_type="language", content="JavaScript programming language", x=100, y=120)
graph.add_node("web_dev", node_type="concept", content="Web Development", x=24, y=30)
graph.add_node("ml", node_type="concept", content="Machine Learning", x=45, y=60)
graph.add_node(
"metformin",
node_type="drug",
content="Metformin",
aliases=["Glucophage"],
confidence="0.97",
tags=["drug", "featured"],
x=22,
y=33,
)
graph.add_node(
"decision_1",
node_type="decision",
content="Approve ML framework",
category="tech",
scenario="Choosing ML framework",
outcome="approved",
confidence="0.9",
reasoning="Best performance",
x=60,
y=80,
)
graph.add_node(
"decision_2",
node_type="decision",
content="Reject legacy stack",
category="tech",
scenario="Choosing ML framework alternative",
outcome="rejected",
confidence="0.4",
reasoning="Outdated",
x=64,
y=86,
)
graph.add_node(
"temporal_node",
node_type="event",
content="Conference talk",
valid_from="2025-01-01T00:00:00",
valid_until="2025-12-31T23:59:59",
x=12,
y=18,
)
graph.add_edge("python", "ml", edge_type="used_in", weight=0.9, color="#58a6ff")
graph.add_edge("javascript", "web_dev", edge_type="used_in", weight=0.8)
graph.add_edge("python", "web_dev", edge_type="used_in", weight=0.5)
graph.add_edge("decision_1", "ml", edge_type="about")
return graph
@pytest.fixture(scope="module")
def client():
session = GraphSession(_build_sample_graph())
app = create_app(session=session)
with TestClient(app) as test_client:
yield test_client
class TestHealthInfo:
def test_root_without_frontend_bundle_shows_diagnostic(self, tmp_path, monkeypatch, client):
from semantica.explorer import app as app_module
package_dir = tmp_path / "semantica"
(package_dir / "explorer").mkdir(parents=True)
class FakeAppPath:
def __init__(self, *_args, **_kwargs):
self.path = package_dir / "explorer" / "app.py"
def resolve(self):
return self.path.resolve()
monkeypatch.setattr(app_module, "Path", FakeAppPath)
response = client.get("/")
assert response.status_code == 200
assert "Explorer UI not available" in response.text
assert "frontend bundle was not found" in response.text
assert "/docs" in response.text
def test_root_serves_built_spa_when_bundle_exists(self, tmp_path, monkeypatch):
from starlette.testclient import TestClient
from semantica.explorer import app as app_module
from semantica.explorer.app import create_app
package_dir = tmp_path / "semantica"
static_dir = package_dir / "static"
static_dir.mkdir(parents=True)
(static_dir / "index.html").write_text(
'<!doctype html><html><body><div id="root"></div><script src="/assets/app.js"></script></body></html>',
encoding="utf-8",
)
class FakeAppPath:
def __init__(self, *_args, **_kwargs):
self.path = package_dir / "explorer" / "app.py"
def resolve(self):
return self.path.resolve()
monkeypatch.setattr(app_module, "Path", FakeAppPath)
with TestClient(create_app()) as test_client:
response = test_client.get("/")
assert response.status_code == 200
assert '<div id="root"></div>' in response.text
assert '<script src="/assets/app.js"></script>' in response.text
def test_health(self, client):
response = client.get("/api/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_info(self, client):
response = client.get("/api/info")
assert response.status_code == 200
payload = response.json()
assert payload["name"] == "Semantica Knowledge Explorer"
assert payload["status"] == "active"
assert payload["version"]
def test_env_settings_are_read_from_supported_names(self, monkeypatch):
monkeypatch.setenv(
"ALLOWED_ORIGINS",
"https://app.example.com, https://team.example.com",
)
monkeypatch.setenv("FALKORDB_HOST", "falkordb.internal")
monkeypatch.setenv("FALKORDB_PORT", "6380")
app = create_app()
assert app.state.explorer_settings["allowed_origins"] == [
"https://app.example.com",
"https://team.example.com",
]
assert app.state.explorer_settings["falkordb_host"] == "falkordb.internal"
assert app.state.explorer_settings["falkordb_port"] == 6380
def test_env_settings_fall_back_to_legacy_cors_name(self, monkeypatch):
monkeypatch.delenv("ALLOWED_ORIGINS", raising=False)
monkeypatch.setenv("EXPLORER_CORS_ORIGINS", "https://legacy.example.com")
app = create_app()
assert app.state.explorer_settings["allowed_origins"] == ["https://legacy.example.com"]
def test_default_app_initializes_empty_graph_session(self):
with TestClient(create_app()) as test_client:
response = test_client.get("/api/graph/nodes")
assert response.status_code == 200
payload = response.json()
assert payload["nodes"] == []
assert payload["total"] == 0
class TestGraphNodes:
def test_list_nodes(self, client):
response = client.get("/api/graph/nodes")
assert response.status_code == 200
payload = response.json()
assert payload["total"] >= 7
assert len(payload["nodes"]) <= payload["total"]
assert payload["has_more"] in {True, False}
def test_list_nodes_filter_type(self, client):
response = client.get("/api/graph/nodes?type=language")
assert response.status_code == 200
assert all(node["type"] == "language" for node in response.json()["nodes"])
def test_list_nodes_search(self, client):
response = client.get("/api/graph/nodes?search=python")
assert response.status_code == 200
payload = response.json()
assert any(node["id"] == "python" for node in payload["nodes"])
assert all(node["properties"].get("content") for node in payload["nodes"])
def test_list_nodes_cursor_pagination(self, client):
first_page = client.get("/api/graph/nodes?limit=2")
assert first_page.status_code == 200
first_payload = first_page.json()
assert len(first_payload["nodes"]) == 2
assert first_payload["next_cursor"]
second_page = client.get(f"/api/graph/nodes?limit=2&cursor={first_payload['next_cursor']}")
assert second_page.status_code == 200
second_payload = second_page.json()
first_ids = {node["id"] for node in first_payload["nodes"]}
second_ids = {node["id"] for node in second_payload["nodes"]}
assert first_ids.isdisjoint(second_ids)
def test_list_nodes_bbox_filter(self, client):
response = client.get("/api/graph/nodes?bbox=0,0,30,40")
assert response.status_code == 200
payload = response.json()
ids = {node["id"] for node in payload["nodes"]}
assert "python" in ids
assert "web_dev" in ids
assert "javascript" not in ids
def test_get_node(self, client):
response = client.get("/api/graph/node/python")
assert response.status_code == 200
payload = response.json()
assert payload["id"] == "python"
assert payload["properties"]["content"] == "Python programming language"
def test_get_neighbors(self, client):
response = client.get("/api/graph/node/python/neighbors?depth=2")
assert response.status_code == 200
payload = response.json()
assert len(payload) >= 1
assert any(item["id"] in {"ml", "web_dev"} for item in payload)
class TestGraphEdges:
def test_list_edges(self, client):
response = client.get("/api/graph/edges")
assert response.status_code == 200
payload = response.json()
assert payload["total"] >= 4
assert all(edge["source"] and edge["target"] and edge["type"] for edge in payload["edges"])
def test_list_edges_filter_source_target(self, client):
response = client.get("/api/graph/edges?source=python&target=ml")
assert response.status_code == 200
payload = response.json()
assert len(payload["edges"]) == 1
assert payload["edges"][0]["type"] == "used_in"
def test_list_edges_cursor_pagination(self, client):
first_page = client.get("/api/graph/edges?limit=2")
assert first_page.status_code == 200
first_payload = first_page.json()
assert len(first_payload["edges"]) == 2
assert first_payload["next_cursor"]
second_page = client.get(f"/api/graph/edges?limit=2&cursor={first_payload['next_cursor']}")
assert second_page.status_code == 200
second_payload = second_page.json()
assert {json.dumps(edge, sort_keys=True) for edge in first_payload["edges"]}.isdisjoint(
{json.dumps(edge, sort_keys=True) for edge in second_payload["edges"]}
)
def test_list_edges_repeated_request_is_stable(self, client):
first_response = client.get("/api/graph/edges?limit=20")
second_response = client.get("/api/graph/edges?limit=20")
assert first_response.status_code == 200
assert second_response.status_code == 200
assert first_response.json() == second_response.json()
def test_list_edges_pagination_union_matches_total(self, client):
cursor = None
seen_ids: set[str] = set()
seen_rows: set[str] = set()
total = None
while True:
url = "/api/graph/edges?limit=2"
if cursor:
url = f"{url}&cursor={cursor}"
response = client.get(url)
assert response.status_code == 200
payload = response.json()
total = payload["total"] if total is None else total
for edge in payload["edges"]:
row_key = json.dumps(edge, sort_keys=True)
assert row_key not in seen_rows
seen_rows.add(row_key)
assert edge["id"] not in seen_ids
seen_ids.add(edge["id"])
cursor = payload["next_cursor"]
if not cursor:
break
assert total is not None
assert len(seen_ids) == total
class TestSearchAndStats:
def test_search(self, client):
response = client.post(
"/api/graph/search",
json={"query": "programming", "filters": {"type": "language"}, "limit": 5},
)
assert response.status_code == 200
payload = response.json()
assert payload["query"] == "programming"
assert payload["total"] >= 1
assert all(item["node"]["type"] == "language" for item in payload["results"])
def test_search_exact_and_prefix(self, client):
exact_response = client.post(
"/api/graph/search",
json={"query": "Metformin", "limit": 5},
)
assert exact_response.status_code == 200
exact_payload = exact_response.json()
assert exact_payload["results"][0]["node"]["id"] == "metformin"
prefix_response = client.post(
"/api/graph/search",
json={"query": "metf", "limit": 5},
)
assert prefix_response.status_code == 200
prefix_payload = prefix_response.json()
assert any(item["node"]["id"] == "metformin" for item in prefix_payload["results"])
def test_search_filters_and_cache_stability(self, client):
body = {
"query": "framework",
"filters": {"type": "decision", "min_confidence": 0.8},
"limit": 5,
}
first_response = client.post("/api/graph/search", json=body)
second_response = client.post("/api/graph/search", json=body)
assert first_response.status_code == 200
assert second_response.status_code == 200
assert first_response.json() == second_response.json()
results = first_response.json()["results"]
assert [item["node"]["id"] for item in results] == ["decision_1"]
def test_search_sees_new_nodes_after_mutation(self, client):
session = client.app.state.session
assert session.add_node(
"metformin_hcl",
"drug",
content="Metformin Hydrochloride",
aliases=["Glucophage XR"],
confidence="0.93",
)
response = client.post(
"/api/graph/search",
json={"query": "glucophage", "limit": 10},
)
assert response.status_code == 200
result_ids = [item["node"]["id"] for item in response.json()["results"]]
assert "metformin" in result_ids
assert "metformin_hcl" in result_ids
def test_search_secondary_scan_fallback_matches_non_curated_properties(self, client):
session = client.app.state.session
assert session.add_node(
"fallback_node",
"entity",
content="Alpha",
description="rareterm",
)
response = client.post(
"/api/graph/search",
json={"query": "rareterm", "limit": 10},
)
assert response.status_code == 200
result_ids = [item["node"]["id"] for item in response.json()["results"]]
assert "fallback_node" in result_ids
def test_stats(self, client):
response = client.get("/api/graph/stats")
assert response.status_code == 200
payload = response.json()
assert payload["node_count"] >= 7
assert payload["edge_count"] >= 4
assert payload["density"] >= 0
class TestDecisions:
def test_list_decisions(self, client):
response = client.get("/api/decisions")
assert response.status_code == 200
payload = response.json()
assert len(payload) >= 2
assert all("decision_id" in item for item in payload)
def test_get_decision(self, client):
response = client.get("/api/decisions/decision_1")
assert response.status_code == 200
payload = response.json()
assert payload["decision_id"] == "decision_1"
assert payload["outcome"] == "approved"
def test_precedents(self, client):
response = client.get("/api/decisions/decision_1/precedents")
assert response.status_code == 200
ids = {item["decision_id"] for item in response.json()}
assert "decision_2" in ids
def test_compliance(self, client):
response = client.get("/api/decisions/decision_1/compliance")
assert response.status_code == 200
assert response.json()["compliant"] is True
client.app.state.session.graph.add_node("policy_1", node_type="policy", content="Data policy")
client.app.state.session.graph.add_edge("decision_1", "policy_1", edge_type="violates")
violation_response = client.get("/api/decisions/decision_1/compliance")
assert violation_response.status_code == 200
assert violation_response.json()["compliant"] is False
@pytest.fixture(scope="module")
def recorded_client():
"""Client over a graph whose decisions were written by record_decision()."""
graph = ContextGraph(advanced_analytics=False)
entities = ["applicant_A7291"]
graph.record_decision(
category="credit_application",
scenario="Personal loan, $85k income, 31% DTI",
reasoning="Income meets threshold; employment stable",
outcome="proceed_to_underwriting",
confidence=0.88,
entities=entities,
)
graph.record_decision(
category="loan_underwriting",
scenario="Underwriting review for A-7291",
reasoning="DTI within policy; clean 36-month credit history",
outcome="approved",
confidence=0.94,
entities=entities,
)
with TestClient(create_app(session=GraphSession(graph))) as test_client:
yield test_client
class TestRecordedDecisions:
"""Decisions written by record_decision(), not hand-built decision nodes.
record_decision() stores ``timestamp`` as a float epoch. The fixtures above
set no timestamp at all, so these routes were only ever exercised against
decision nodes that could not trigger the float/str mismatch.
"""
def test_list_decisions_serializes_float_timestamp(self, recorded_client):
response = recorded_client.get("/api/decisions")
assert response.status_code == 200
payload = response.json()
assert len(payload) == 2
for item in payload:
assert isinstance(item["timestamp"], str)
datetime.fromisoformat(item["timestamp"])
def test_get_decision(self, recorded_client):
listed = recorded_client.get("/api/decisions").json()
decision_id = listed[0]["decision_id"]
response = recorded_client.get(f"/api/decisions/{decision_id}")
assert response.status_code == 200
assert response.json()["decision_id"] == decision_id
def test_filter_by_category(self, recorded_client):
response = recorded_client.get("/api/decisions?category=loan_underwriting")
assert response.status_code == 200
payload = response.json()
assert len(payload) == 1
assert payload[0]["outcome"] == "approved"
class TestTemporal:
def test_snapshot_now(self, client):
response = client.get("/api/temporal/snapshot")
assert response.status_code == 200
payload = response.json()
assert payload["active_node_count"] >= 1
assert isinstance(payload["active_node_ids"], list)
def test_snapshot_at(self, client):
active_response = client.get("/api/temporal/snapshot?at=2025-06-15T00:00:00")
assert active_response.status_code == 200
assert "temporal_node" in active_response.json()["active_node_ids"]
inactive_response = client.get("/api/temporal/snapshot?at=2026-01-01T00:00:00")
assert inactive_response.status_code == 200
assert "temporal_node" not in inactive_response.json()["active_node_ids"]
def test_diff(self, client):
response = client.get(
"/api/temporal/diff?from_time=2024-01-01T00:00:00&to_time=2025-06-15T00:00:00"
)
assert response.status_code == 200
payload = response.json()
assert "temporal_node" in payload["added_nodes"]
def test_patterns(self, client):
response = client.get("/api/temporal/patterns")
assert response.status_code == 200
assert "patterns" in response.json()
def test_patterns_failure_returns_500(self, client, monkeypatch):
session = client.app.state.session
def _boom():
raise RuntimeError("boom")
monkeypatch.setattr(session, "build_graph_dict", _boom)
response = client.get("/api/temporal/patterns")
assert response.status_code == 500
def test_bounds(self, client):
response = client.get("/api/temporal/bounds")
assert response.status_code == 200
payload = response.json()
assert "min" in payload
assert "max" in payload
class TestAnalytics:
def test_analytics(self, client):
response = client.get("/api/analytics?metrics=centrality")
assert response.status_code == 200
assert "centrality" in response.json()
def test_analytics_partial_failure_returns_207(self, client, monkeypatch):
session = client.app.state.session
def _boom(*_args, **_kwargs):
raise RuntimeError("boom")
monkeypatch.setattr(session.centrality, "calculate_degree_centrality", _boom)
response = client.get("/api/analytics?metrics=centrality,community")
assert response.status_code == 207
payload = response.json()
assert payload["centrality"]["error"]
assert payload["community"] is not None and "error" not in payload["community"]
def test_analytics_total_failure_returns_500(self, client, monkeypatch):
session = client.app.state.session
def _boom(*_args, **_kwargs):
raise RuntimeError("boom")
monkeypatch.setattr(session.centrality, "calculate_degree_centrality", _boom)
response = client.get("/api/analytics?metrics=centrality")
assert response.status_code == 500
def test_validation(self, client):
response = client.get("/api/analytics/validation")
assert response.status_code == 200
payload = response.json()
assert "valid" in payload
assert "issues" in payload
class TestOntologyCreateFailures:
def test_create_from_sample_data_failure_returns_500(self, client, monkeypatch):
from semantica.ontology import OntologyEngine
def _boom(self, *_args, **_kwargs):
raise RuntimeError("boom")
monkeypatch.setattr(OntologyEngine, "from_data", _boom)
response = client.post(
"/api/ontology/create",
json={
"mode": "data",
"namespace": "http://example.org/create-failure-data",
"name": "Create Failure (data)",
"sample_data": "id,name\n1,Alice\n2,Bob",
},
)
assert response.status_code == 500
def test_create_from_text_failure_returns_500(self, client, monkeypatch):
from semantica.ontology import OntologyEngine
def _boom(self, *_args, **_kwargs):
raise RuntimeError("boom")
monkeypatch.setattr(OntologyEngine, "from_text", _boom)
response = client.post(
"/api/ontology/create",
json={
"mode": "text",
"namespace": "http://example.org/create-failure-text",
"name": "Create Failure (text)",
"schema_text": "Class: Person\nProperty: knows",
},
)
assert response.status_code == 500
class TestEnrichment:
def test_reasoning(self, client):
response = client.post(
"/api/reason",
json={
"facts": ["Person(Alice)", "Knows(Alice, Bob)"],
"rules": ["IF Knows(?x, ?y) THEN Connected(?x, ?y)"],
"mode": "forward",
},
)
assert response.status_code in (200, 422)
def test_reasoning_apply_to_graph_fallback(self, client):
response = client.post(
"/api/reason",
json={
"facts": ["inhibits(Metformin, mTOR)", "causes(mTOR, Neurodegeneration)"],
"rules": [
"IF inhibits(Metformin, mTOR) AND causes(mTOR, Neurodegeneration) THEN candidate(Metformin, Alzheimer's)"
],
"mode": "forward",
"apply_to_graph": True,
},
)
assert response.status_code == 200
payload = response.json()
assert "candidate(Metformin, Alzheimer's)" in payload["inferred_facts"]
assert payload["added_edges"] >= 1
edge_lookup = client.get("/api/graph/edges?source=Metformin&target=Alzheimer%27s")
assert edge_lookup.status_code == 200
assert edge_lookup.json()["edges"][0]["properties"]["inferred"] is True
def test_extract(self, client):
response = client.post("/api/enrich/extract", json={"text": "Alice works at Acme Corp."})
# 503 is reserved for a genuinely absent semantic_extract module; it must
# not be reachable on an install where the module imports cleanly.
# Runtime errors from the extraction stack surface as 500, not 422.
assert response.status_code in (200, 422, 500)
def test_extract_returns_entities(self, client):
response = client.post(
"/api/enrich/extract",
json={"text": "Apple CEO Tim Cook announced record earnings in Cupertino."},
)
assert response.status_code == 200
payload = response.json()
assert payload["entities"], "extraction returned no entities"
assert any("Tim Cook" in str(entity) for entity in payload["entities"])
def test_link_prediction(self, client):
response = client.post("/api/enrich/links", json={"node_id": "python", "top_n": 5})
assert response.status_code in (200, 422)
def test_dedup(self, client):
response = client.post("/api/enrich/dedup", json={"threshold": 0.8})
assert response.status_code in (200, 422)
class TestAnnotations:
def test_create_list_delete(self, client):
created = client.post(
"/api/annotations",
json={"node_id": "python", "content": "Great language!", "tags": ["fav"]},
)
assert created.status_code == 201
annotation = created.json()
annotation_id = annotation["annotation_id"]
listed = client.get("/api/annotations?node_id=python")
assert listed.status_code == 200
assert any(item["annotation_id"] == annotation_id for item in listed.json())
deleted = client.delete(f"/api/annotations/{annotation_id}")
assert deleted.status_code == 204
class TestImportExport:
def test_export_json(self, client):
response = client.post("/api/export", json={"format": "json"})
assert response.status_code == 200
payload = response.json()
assert "entities" in payload
assert "relationships" in payload
def test_export_csv(self, client):
response = client.post("/api/export", json={"format": "csv"})
assert response.status_code == 200
assert "text/csv" in response.headers["content-type"].lower()
@pytest.mark.parametrize(
"fmt,rdflib_format",
[
("turtle", "turtle"),
("ttl", "turtle"),
("nt", "nt"),
("ntriples", "nt"),
("n-triples", "nt"),
("xml", "xml"),
("rdfxml", "xml"),
("rdf-xml", "xml"),
("jsonld", "json-ld"),
("json-ld", "json-ld"),
],
)
def test_export_rdf_formats(self, client, fmt, rdflib_format):
"""The Explorer used to answer 422 for every RDF format while the MCP
`export_graph` tool offered them, so a graph could be loaded as JSON-LD and never
exported back (#1131).
Parsed with a real RDF parser rather than asserted on strings: a response that
merely *looks* like Turtle is what makes this class of gap survive a test suite.
"""
rdflib = pytest.importorskip("rdflib")
response = client.post("/api/export", json={"format": fmt})
assert response.status_code == 200, response.text
graph = rdflib.Graph()
graph.parse(data=response.text, format=rdflib_format)
assert len(graph) > 0, f"{fmt} export parsed to zero triples"
def test_export_aliases_agree_with_mcp_tool_where_overlapping(self):
"""The two surfaces of one product should not disagree about what `ttl` means.
Explorer now maps to RDFExporter canonical formats (e.g., nt->ntriples),
while MCP maps to its own intermediates (e.g., nt->nt). This test verifies
that where MCP and Explorer overlap in alias names, they ultimately work
correctly even if the intermediate canonical form differs.
Canary: if either alias table drifts such that an alias becomes unsupported,
this test will catch it."""
from mcp.tools.export import _FORMAT_ALIASES as MCP_ALIASES
from semantica.explorer.routes.export_import import _RDF_FORMATS
# Verify all MCP aliases are present in Explorer
for alias in MCP_ALIASES.keys():
assert alias in _RDF_FORMATS, (
f"MCP alias {alias!r} not present in Explorer _RDF_FORMATS"
)
# Note: We don't require identical canonical forms because:
# - MCP maps to intermediates that RDFExporter then translates
# - Explorer now maps directly to RDFExporter canonical forms
# - Both ultimately work correctly
def test_export_graphml(self, client):
"""GraphML export should work using GraphExporter."""
response = client.post("/api/export", json={"format": "graphml"})
assert response.status_code == 200, response.text
assert "application/xml" in response.headers["content-type"].lower()
# Verify it's valid XML and contains GraphML structure
content = response.text
assert '<?xml version="1.0"' in content
assert '<graphml' in content
assert '</graphml>' in content
def test_export_empty_graph_rdf(self, client):
"""Empty graphs should export successfully in RDF formats."""
# First, clear the graph or use a clean client
# This test assumes test fixtures provide a graph; for empty graph
# we'd need to manipulate the session, which may not be straightforward
# in these integration tests. Keeping this as documentation.
pass
def test_export_rdf_validation_error_handling(self, client):
"""RDF validation errors should return HTTP 422, not 500."""
# This would require crafting malformed graph data that passes
# session.build_graph_dict() but fails RDF validation.
# Since build_graph_dict() returns valid structure, this is difficult
# to trigger in integration tests. Keeping as documentation.
pass
def test_unsupported_format_names_what_is_supported(self, client):
"""The old message said only that the format was unsupported, which reads as 'this
format does not exist' rather than 'this door does not open it'."""
response = client.post("/api/export", json={"format": "no-such-format"})
assert response.status_code == 422
detail = response.json()["detail"]
assert "turtle" in detail and "json" in detail
def test_import_json_with_edge_metadata(self, client):
payload = json.dumps(
{
"nodes": [
{"id": "meta_src", "type": "test", "properties": {"content": "src"}},
{"id": "meta_tgt", "type": "test", "properties": {"content": "tgt"}},
],
"edges": [
{
"id": "meta-edge-1",
"familyId": "meta-family",
"source": "meta_src",
"target": "meta_tgt",
"type": "tagged",
"metadata": {"label": "important", "weight": 0.7},
}
],
}
)
response = client.post(
"/api/import",
files={"file": ("graph.json", payload, "application/json")},
)
assert response.status_code == 200
body = response.json()
assert body["status"] == "success"
assert body["nodes_added"] == body["nodes_imported"]
assert body["edges_added"] == body["edges_imported"]
edge_lookup = client.get("/api/graph/edges?source=meta_src&target=meta_tgt")
assert edge_lookup.status_code == 200
edge_payload = edge_lookup.json()["edges"][0]
props = edge_payload["properties"]
assert edge_payload["id"] == "meta-edge-1"
assert edge_payload["familyId"] == "meta-family"
assert props["label"] == "important"
def test_import_csv(self, client):
payload = "id,type,content\nnode_csv,entity,Hello CSV\n"
response = client.post(
"/api/import",
files={"file": ("graph.csv", payload, "text/csv")},
)
assert response.status_code == 200
assert response.json()["nodes_added"] >= 1
def test_provenance_report_json(self, client):
response = client.get("/api/provenance/report?node_id=python&format=json")
assert response.status_code == 200
payload = response.json()
assert payload["node_id"] == "python"
assert "lineage" in payload
def test_provenance_report_markdown(self, client):
response = client.get("/api/provenance/report?node_id=python&format=markdown")
assert response.status_code == 200
assert "text/plain" in response.headers["content-type"].lower()
assert "Provenance Report" in response.text
class TestRealtimeUpdates:
def test_websocket_receives_graph_mutation(self, client):
with client.websocket_connect("/ws/graph-updates") as websocket:
ack = websocket.receive_json()
assert ack["event"] == "connection_ack"
client.app.state.session.graph.add_node("ws_node", node_type="entity", content="WebSocket Node")
event = websocket.receive_json()
assert event["event"] == "graph_mutation"
assert event["data"]["event_type"] == "ADD_NODE"
assert event["data"]["entity_id"] == "ws_node"
class TestGenericGraphFileLoading:
@staticmethod
def _write_graph_payload(payload):
tmp_dir = Path("tests") / "explorer" / ".tmp"
tmp_dir.mkdir(parents=True, exist_ok=True)
graph_path = tmp_dir / f"{uuid.uuid4().hex}.json"
graph_path.write_text(json.dumps(payload), encoding="utf-8")
return graph_path
def test_file_loader_accepts_label_and_source_target_shape(self):
payload = {
"metadata": {"dataset": "demo"},
"nodes": [
{"id": "drug::metformin", "type": "drug", "label": "Metformin", "properties": {"source": "PrimeKG"}},
{"id": "gene::mtor", "type": "gene", "label": "mTOR", "properties": {"source": "NCBI"}},
],
"edges": [
{
"id": "drug::metformin::inhibits::gene::mtor",
"source": "drug::metformin",
"target": "gene::mtor",
"type": "inhibits",
"label": "inhibits",
"properties": {"confidence": 0.92},
}
],
}
graph_path = self._write_graph_payload(payload)
session = GraphSession.from_file(str(graph_path))
assert None not in session.graph.nodes
metformin = session.get_node("drug::metformin")
assert metformin is not None
assert metformin["content"] == "Metformin"
edges, total = session.get_edges(limit=10)
assert total == 1
assert edges[0]["source"] == "drug::metformin"
assert edges[0]["target"] == "gene::mtor"
def test_generic_file_loading_keeps_temporal_endpoints_stable(self):
payload = {
"nodes": [
{
"id": "drug::metformin",
"type": "drug",
"label": "Metformin",
"properties": {
"valid_from": "2020-01-01T00:00:00",
"valid_until": "2024-12-31T23:59:59",
},
},
{"id": "disease::alz", "type": "disease", "label": "Alzheimer disease"},
],
"edges": [
{"source": "drug::metformin", "target": "disease::alz", "type": "candidate"},
{"target": "disease::alz", "type": "broken_edge_should_be_ignored"},
],
}
graph_path = self._write_graph_payload(payload)
session = GraphSession.from_file(str(graph_path))
app = create_app(session=session)
with TestClient(app) as test_client:
bounds = test_client.get("/api/temporal/bounds")
assert bounds.status_code == 200
assert bounds.json()["min"] == "2020-01-01T00:00:00"
edges = test_client.get("/api/graph/edges?limit=10")
assert edges.status_code == 200
payload = edges.json()
assert payload["total"] == 1
assert payload["edges"][0]["type"] == "candidate"
def test_generic_file_loading_assigns_stable_legacy_edge_ids(self):
payload = {
"nodes": [
{"id": "legacy_src", "type": "entity", "properties": {"content": "Legacy source"}},
{"id": "legacy_tgt", "type": "entity", "properties": {"content": "Legacy target"}},
],
"edges": [
{
"source": "legacy_src",
"target": "legacy_tgt",
"type": "related_to",
"weight": 0.75,
"metadata": {"evidence": "legacy"},
}
],
}
graph_path = self._write_graph_payload(payload)
session_one = GraphSession.from_file(str(graph_path))
session_two = GraphSession.from_file(str(graph_path))
edges_one, total_one = session_one.get_edges(limit=10)
edges_two, total_two = session_two.get_edges(limit=10)
assert total_one == 1
assert total_two == 1
assert edges_one[0]["id"] == edges_two[0]["id"]
assert edges_one[0]["familyId"] == edges_two[0]["familyId"]
def test_multi_edges_same_pair_paginate_as_distinct_stable_edges(self):
graph = ContextGraph(advanced_analytics=False)
graph.add_node("shared_src", node_type="entity", content="Shared source")
graph.add_node("shared_tgt", node_type="entity", content="Shared target")
graph.add_edges(
[
{
"id": "edge-alpha",
"familyId": "family-shared",
"source_id": "shared_src",
"target_id": "shared_tgt",
"type": "supports",
"weight": 0.9,
"properties": {"provenance": "paper-a"},
},
{
"id": "edge-beta",
"familyId": "family-shared",
"source_id": "shared_src",
"target_id": "shared_tgt",
"type": "contradicts",
"weight": 0.6,
"properties": {"provenance": "paper-b"},
},
]
)
app = create_app(session=GraphSession(graph))
with TestClient(app) as test_client:
first = test_client.get("/api/graph/edges?source=shared_src&target=shared_tgt&limit=1")
assert first.status_code == 200
first_payload = first.json()
assert first_payload["total"] == 2
assert len(first_payload["edges"]) == 1
assert first_payload["edges"][0]["familyId"] == "family-shared"
second = test_client.get(
f"/api/graph/edges?source=shared_src&target=shared_tgt&limit=1&cursor={first_payload['next_cursor']}"
)
assert second.status_code == 200
second_payload = second.json()
assert len(second_payload["edges"]) == 1
assert first_payload["edges"][0]["id"] != second_payload["edges"][0]["id"]
repeat = test_client.get("/api/graph/edges?source=shared_src&target=shared_tgt&limit=10")
assert repeat.status_code == 200
repeat_ids = [edge["id"] for edge in repeat.json()["edges"]]
assert repeat_ids == ["edge-alpha", "edge-beta"]
# ---------------------------------------------------------------------------
# Bidirectional path-finding tests (issue #469)
# ---------------------------------------------------------------------------
def _make_path_session() -> GraphSession:
"""Return a GraphSession whose build_graph_dict yields an nx.DiGraph with A→B only.
GraphSession wraps a ContextGraph (required by create_app), but we patch
build_graph_dict so PathFinder receives an actual NetworkX DiGraph — the
graph type the Explorer is designed to traverse for path queries.
"""
cg = ContextGraph(advanced_analytics=False)
cg.add_node("A", node_type="entity", content="Node A")
cg.add_node("B", node_type="entity", content="Node B")
cg.add_node("gene/protein:6164", node_type="gene/protein", content="RPL34")
cg.add_node("disease/term:1", node_type="disease", content="Slash target")
cg.add_edge("A", "B", edge_type="connects")
cg.add_edge("gene/protein:6164", "disease/term:1", edge_type="connects")
session = GraphSession(cg)
# Patch build_graph_dict to return the directed NetworkX graph that
# PathFinder needs. The ContextGraph dict format is not traversable by
# PathFinder; this mimics how a KG-backed session would expose the graph.
digraph = nx.DiGraph()
digraph.add_edge("A", "B")
digraph.add_edge("gene/protein:6164", "disease/term:1")
session.build_graph_dict = lambda node_ids=None: digraph # type: ignore[method-assign]
return session
@pytest.fixture
def path_client():
session = _make_path_session()
app = create_app(session=session)
with TestClient(app) as c:
yield c
class TestBidirectionalPathRoute:
"""API-level tests for directed=true/false on GET /api/graph/node/{id}/path."""
# ------------------------------------------------------------------
# directed=true (default) — existing directed-only behaviour
# ------------------------------------------------------------------
def test_directed_true_forward_path_found(self, path_client):
"""A→B exists: forward query with directed=true must succeed."""
resp = path_client.get("/api/graph/node/A/path?target=B&directed=true")
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["A", "B"]
assert body["directed"] is True
def test_directed_true_reverse_returns_404(self, path_client):
"""Only A→B exists: reverse query with directed=true must return 404."""
resp = path_client.get("/api/graph/node/B/path?target=A&directed=true")
assert resp.status_code == 404
def test_default_param_reverse_returns_404(self, path_client):
"""Omitting directed= must preserve current directed behaviour (404 for reverse)."""
resp = path_client.get("/api/graph/node/B/path?target=A")
assert resp.status_code == 404
# ------------------------------------------------------------------
# directed=false — new undirected traversal
# ------------------------------------------------------------------
def test_directed_false_reverse_path_found(self, path_client):
"""directed=false must find B→A even though only A→B exists."""
resp = path_client.get("/api/graph/node/B/path?target=A&directed=false")
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["B", "A"]
assert body["directed"] is False
def test_query_path_route_supports_slash_node_ids(self, path_client):
"""Query-param path route must support arbitrary graph ids with slashes."""
resp = path_client.get(
"/api/graph/path",
params={
"source": "gene/protein:6164",
"target": "disease/term:1",
"algorithm": "dijkstra",
},
)
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["gene/protein:6164", "disease/term:1"]
assert body["source"] == "gene/protein:6164"
assert body["target"] == "disease/term:1"
def test_directed_false_forward_path_found(self, path_client):
"""directed=false must not break the natural A→B direction."""
resp = path_client.get("/api/graph/node/A/path?target=B&directed=false")
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["A", "B"]
assert body["directed"] is False
# ------------------------------------------------------------------
# Algorithm variants
# ------------------------------------------------------------------
def test_dijkstra_directed_false_reverse(self, path_client):
resp = path_client.get(
"/api/graph/node/B/path?target=A&algorithm=dijkstra&directed=false"
)
assert resp.status_code == 200
body = resp.json()
assert body["path"] == ["B", "A"]
assert body["algorithm"] == "dijkstra"
assert body["directed"] is False
def test_dijkstra_directed_true_reverse_returns_404(self, path_client):
resp = path_client.get(
"/api/graph/node/B/path?target=A&algorithm=dijkstra&directed=true"
)
assert resp.status_code == 404
# ------------------------------------------------------------------
# PathResponse schema
# ------------------------------------------------------------------
def test_response_schema_includes_directed_field(self, path_client):
"""PathResponse must always include the directed field."""
resp = path_client.get("/api/graph/node/A/path?target=B")
assert resp.status_code == 200
body = resp.json()
assert "directed" in body
def test_response_directed_reflects_query_param(self, path_client):
resp_true = path_client.get("/api/graph/node/A/path?target=B&directed=true")
resp_false = path_client.get("/api/graph/node/A/path?target=B&directed=false")
assert resp_true.json()["directed"] is True
assert resp_false.json()["directed"] is False
# ------------------------------------------------------------------
# hop_count and distance_band — issue #472
# ------------------------------------------------------------------
def test_response_includes_hop_count_and_distance_band(self, path_client):
"""PathResponse must include hop_count and distance_band fields."""
resp = path_client.get("/api/graph/node/A/path?target=B")
assert resp.status_code == 200
body = resp.json()
assert "hop_count" in body
assert "distance_band" in body
def test_one_hop_path_is_direct(self, path_client):
"""A single-edge path (1 hop) must return distance_band='direct'."""
resp = path_client.get("/api/graph/node/A/path?target=B")
assert resp.status_code == 200
body = resp.json()
assert body["hop_count"] == 1
assert body["distance_band"] == "direct"
# ---------------------------------------------------------------------------
# _classify_distance unit tests — issue #472
# ---------------------------------------------------------------------------
from semantica.utils.helpers import classify_path_distance # noqa: E402
class _FakeSimilarity:
"""Minimal similarity stub shared by slash-safe distance route tests.
Expects embeddings keyed on 'gene/protein:6164' with query vector [1, 0, 0]
and returns a single neighbor result. Tests that need different behaviour
can assign a lambda to instance.find_most_similar after construction.
"""
def find_most_similar(self, embeddings, query_embedding, top_k=10):
assert "gene/protein:6164" in embeddings
assert query_embedding == [1.0, 0.0, 0.0]
return [("disease/term:1", 0.74)]
def _make_slash_node_session(*, with_embeddings: bool = True) -> GraphSession:
"""Return an isolated GraphSession with slash-containing node IDs."""
graph = ContextGraph(advanced_analytics=False)
kwargs = {"embedding": [1.0, 0.0, 0.0]} if with_embeddings else {}
graph.add_node("gene/protein:6164", node_type="gene/protein", content="RPL34", **kwargs)
graph.add_node(
"disease/term:1",
node_type="disease",
content="Slash target",
**({"embedding": [0.7, 0.2, 0.1]} if with_embeddings else {}),
)
session = GraphSession(graph)
session._similarity = _FakeSimilarity()
return session
class TestSlashSafeDistanceRoutes:
def test_query_semantic_neighborhood_supports_slash_node_ids(self):
session = _make_slash_node_session(with_embeddings=True)
app = create_app(session=session)
with TestClient(app) as test_client:
resp = test_client.get(
"/api/graph/semantic-neighborhood",
params={"node_id": "gene/protein:6164", "top_k": 50},
)
assert resp.status_code == 200
body = resp.json()
assert body["anchor_node"] == "gene/protein:6164"
assert body["neighbors"][0]["id"] == "disease/term:1"
assert body["neighbors"][0]["similarity"] == 0.74
def test_legacy_semantic_neighborhood_still_works_for_simple_ids(self):
"""Legacy path-segment route must still return 200 for slash-free node IDs."""
graph = ContextGraph(advanced_analytics=False)
graph.add_node(
"semantic_anchor",
node_type="entity",
content="Semantic anchor",
embedding=[1.0, 0.0, 0.0],
)
graph.add_node(
"semantic_neighbor",
node_type="entity",
content="Semantic neighbor",
embedding=[0.8, 0.2, 0.0],
)
session = GraphSession(graph)
fake = _FakeSimilarity()
fake.find_most_similar = (
lambda embeddings, query_embedding, top_k=10: [("semantic_neighbor", 0.8)]
)
session._similarity = fake
app = create_app(session=session)
with TestClient(app) as test_client:
resp = test_client.get(
"/api/graph/node/semantic_anchor/semantic-neighborhood?top_k=10"
)
assert resp.status_code == 200
assert resp.json()["anchor_node"] == "semantic_anchor"
def test_query_semantic_neighborhood_missing_node_returns_404(self, client):
resp = client.get(
"/api/graph/semantic-neighborhood",
params={"node_id": "gene/protein:missing"},
)
assert resp.status_code == 404
def test_query_semantic_neighborhood_without_embeddings_returns_503(self):
session = _make_slash_node_session(with_embeddings=False)
app = create_app(session=session)
with TestClient(app) as test_client:
resp = test_client.get(
"/api/graph/semantic-neighborhood",
params={"node_id": "gene/protein:6164", "top_k": 50},
)
assert resp.status_code == 503
class TestClassifyDistance:
"""Unit tests covering all four band boundaries."""
def test_zero_hops_is_direct(self):
assert classify_path_distance(0) == "direct"
def test_one_hop_is_direct(self):
assert classify_path_distance(1) == "direct"
def test_two_hops_is_near(self):
assert classify_path_distance(2) == "near"
def test_three_hops_is_near(self):
assert classify_path_distance(3) == "near"
def test_four_hops_is_mid_range(self):
assert classify_path_distance(4) == "mid-range"
def test_six_hops_is_mid_range(self):
assert classify_path_distance(6) == "mid-range"
def test_seven_hops_is_distant(self):
assert classify_path_distance(7) == "distant"
def test_large_hop_count_is_distant(self):
assert classify_path_distance(20) == "distant"
# ---------------------------------------------------------------------------
# Timestamp validator unit tests (no HTTP server needed)
# ---------------------------------------------------------------------------
class TestDecisionResponseTimestampValidator:
"""Unit tests for DecisionResponse._normalize_timestamp.
These run directly against the Pydantic model, not through the HTTP stack,
so they are fast and isolated from the rest of the Explorer infrastructure.
"""
def _make(self, ts):
from semantica.explorer.schemas import DecisionResponse
import pytest as _pytest
return DecisionResponse(decision_id="x", timestamp=ts)
def test_none_passes_through(self):
from semantica.explorer.schemas import DecisionResponse
dr = DecisionResponse(decision_id="x", timestamp=None)
assert dr.timestamp is None
def test_string_passes_through_unchanged(self):
from semantica.explorer.schemas import DecisionResponse
iso = "2024-08-14T10:23:45+00:00"
dr = DecisionResponse(decision_id="x", timestamp=iso)
assert dr.timestamp == iso
def test_float_epoch_becomes_iso_string(self):
from datetime import datetime, timezone
from semantica.explorer.schemas import DecisionResponse
epoch = 1723600000.5
dr = DecisionResponse(decision_id="x", timestamp=epoch)
assert isinstance(dr.timestamp, str)
parsed = datetime.fromisoformat(dr.timestamp)
assert abs(parsed.timestamp() - epoch) < 1.0
def test_int_epoch_becomes_iso_string(self):
from datetime import datetime
from semantica.explorer.schemas import DecisionResponse
epoch = 1723600000
dr = DecisionResponse(decision_id="x", timestamp=epoch)
assert isinstance(dr.timestamp, str)
datetime.fromisoformat(dr.timestamp)
def test_nan_raises_validation_error(self):
import math
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp=math.nan)
def test_positive_inf_raises_validation_error(self):
import math
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp=math.inf)
def test_negative_inf_raises_validation_error(self):
import math
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp=-math.inf)
def test_dict_raises_validation_error(self):
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp={"$date": 1723600000})
def test_list_raises_validation_error(self):
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp=[1723600000])
def test_bool_raises_validation_error(self):
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp=True)
def test_oserror_range_epoch_raises_validation_error(self):
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
# Milliseconds mistakenly stored where seconds were expected.
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp=1723600000000)
def test_overflow_range_epoch_raises_validation_error(self):
import pytest
from pydantic import ValidationError
from semantica.explorer.schemas import DecisionResponse
with pytest.raises(ValidationError):
DecisionResponse(decision_id="x", timestamp=1e20)
# ---------------------------------------------------------------------------
# /api/enrich/extract input-size and import-boundary tests
# ---------------------------------------------------------------------------
class TestEnrichExtractValidation:
"""Tests for the input constraints and exception handling added to
POST /api/enrich/extract."""
def test_oversized_input_rejected_before_nlp(self, client):
"""A payload exceeding the 10 000-character limit must be rejected with
422 before any NLP work is attempted."""
oversized = "a " * 5_001 # 10 002 characters
response = client.post("/api/enrich/extract", json={"text": oversized})
assert response.status_code == 422
def test_input_at_limit_is_accepted(self, client):
"""A payload at exactly the maximum length must not be rejected by the
schema validator (NLP may still fail, but the schema must accept it)."""
at_limit = "a" * 10_000
response = client.post("/api/enrich/extract", json={"text": at_limit})
# 503 = module missing, 500 = runtime error from the extraction stack,
# 200 = success. What must NOT happen is a schema rejection (422 from
# Pydantic due to max_length), since this input is exactly at the limit.
assert response.status_code in (200, 500, 503)
def test_import_failure_returns_503_not_422(self, client, monkeypatch):
"""A genuine ImportError on the semantic_extract import must produce 503
(dependency unavailable), NOT 422 (extraction failed)."""
import semantica.explorer.routes.enrich as enrich_module
def _failing_import(name, *args, **kwargs):
if "semantic_extract" in name:
raise ImportError("semantic_extract not installed")
return original_import(name, *args, **kwargs)
import builtins
original_import = builtins.__import__
monkeypatch.setattr(builtins, "__import__", _failing_import)
response = client.post(
"/api/enrich/extract",
json={"text": "Apple was founded by Steve Jobs."},
)
assert response.status_code == 503
assert "semantic_extract" in response.json()["detail"].lower()
Reference in New Issue View Git Blame Copy Permalink