mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
521 lines
20 KiB
Python
521 lines
20 KiB
Python
"""Integration tests for the explorer API."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import uuid
|
|
|
|
import pytest
|
|
|
|
from semantica.context.context_graph import ContextGraph
|
|
from semantica.explorer.app import create_app
|
|
from semantica.explorer.session import GraphSession
|
|
|
|
try:
|
|
from starlette.testclient import TestClient
|
|
except ImportError:
|
|
pytest.skip(
|
|
"starlette TestClient is required for explorer tests. Install semantica[explorer].",
|
|
allow_module_level=True,
|
|
)
|
|
|
|
|
|
|
|
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(
|
|
"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_serves_spa(self, client):
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert '<div id="root"></div>' in response.text
|
|
|
|
def test_health(self, client):
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "healthy"
|
|
|
|
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"]
|
|
|
|
|
|
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"]}
|
|
)
|
|
|
|
|
|
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_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
|
|
|
|
|
|
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_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_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 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."})
|
|
assert response.status_code in (200, 422)
|
|
|
|
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()
|
|
|
|
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": [
|
|
{
|
|
"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
|
|
props = edge_lookup.json()["edges"][0]["properties"]
|
|
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"
|