mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-08-29 04:26:20 +00:00
KaifAhmad1andmikemikimike
bc75768afe
- validate_skos_hierarchy() re-walked every existing hierarchy edge in the graph on each write, so one pre-existing cycle anywhere would block all unrelated future SKOS writes. It now only traverses concepts touched by the edges actually being written, while still checking those against existing edges for cross-boundary cycles. - In /api/ontology/load, `except HTTPException: raise` sat after a broader `except Exception` clause that already matched HTTPException, so a 422 raised after a successful OntologyIngestor parse was silently swallowed and retried via the fallback RDF parser instead of reaching the caller. Reordered the except clauses. Co-authored-by: mikemikimike <13286568797@163.com>
257 lines
9.9 KiB
Python
257 lines
9.9 KiB
Python
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
from datetime import datetime
|
|
import sys
|
|
import os
|
|
|
|
# Ensure the semantica package is in the path
|
|
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
|
|
|
|
from semantica.context.entity_linker import EntityLinker, LinkedEntity, EntityLink
|
|
from semantica.context.context_graph import ContextGraph, ContextNode, ContextEdge
|
|
from semantica.context.agent_memory import AgentMemory, MemoryItem
|
|
from semantica.context.context_retriever import ContextRetriever, RetrievedContext
|
|
from semantica.context.agent_context import AgentContext
|
|
|
|
class MockVectorStore:
|
|
def __init__(self):
|
|
self.vectors = []
|
|
self.metadata = []
|
|
|
|
def store_vectors(self, vectors, metadata):
|
|
self.vectors.extend(vectors)
|
|
self.metadata.extend(metadata)
|
|
|
|
def add(self, items):
|
|
# Support add protocol
|
|
for item in items:
|
|
self.metadata.append(item.metadata)
|
|
|
|
def search(self, query_vector, k=5):
|
|
# Mock search return
|
|
return []
|
|
|
|
class TestContextModule(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.mock_vector_store = MockVectorStore()
|
|
self.mock_kg = MagicMock()
|
|
|
|
# --- EntityLinker Tests ---
|
|
def test_entity_linker_assign_uri(self):
|
|
linker = EntityLinker(base_uri="http://example.com/")
|
|
|
|
# Test text-based URI
|
|
uri1 = linker.assign_uri("id1", "Test Entity", "TEST")
|
|
self.assertEqual(uri1, "http://example.com/test_entity#test")
|
|
|
|
# Test hash-based URI
|
|
uri2 = linker.assign_uri("id2")
|
|
self.assertTrue(uri2.startswith("http://example.com/"))
|
|
|
|
# Test registry
|
|
uri3 = linker.assign_uri("id1")
|
|
self.assertEqual(uri3, uri1)
|
|
|
|
def test_entity_linker_link(self):
|
|
linker = EntityLinker()
|
|
entities = [{"text": "Python", "label": "LANGUAGE", "start": 0, "end": 6}]
|
|
linked = linker.link("Python code", entities=entities)
|
|
# Note: The current implementation of link might be a placeholder or depend on logic
|
|
# that returns empty if no detailed logic is implemented.
|
|
# Based on my read, it tracks progress but might not implement full logic without external NLP.
|
|
# However, checking it runs without error is a good start.
|
|
self.assertIsInstance(linked, list)
|
|
|
|
def test_entity_linker_find_similar_entities_returns_dicts(self):
|
|
linker = EntityLinker(
|
|
knowledge_graph={
|
|
"entities": [
|
|
{"id": "lang_python", "text": "Python programming language", "type": "Technology"}
|
|
]
|
|
}
|
|
)
|
|
linker.assign_uri("lang_python", "Python programming language", "Technology")
|
|
|
|
similar = linker.find_similar_entities("Python programming language", threshold=0.5)
|
|
|
|
self.assertEqual(len(similar), 1)
|
|
self.assertEqual(similar[0]["entity_id"], "lang_python")
|
|
self.assertEqual(similar[0]["text"], "Python programming language")
|
|
self.assertIn("similarity", similar[0])
|
|
|
|
# --- ContextGraph Tests ---
|
|
def test_context_graph_operations(self):
|
|
graph = ContextGraph()
|
|
|
|
# Add nodes
|
|
nodes = [
|
|
{"id": "n1", "type": "person", "properties": {"name": "Alice"}},
|
|
{"id": "n2", "type": "person", "properties": {"name": "Bob"}}
|
|
]
|
|
count = graph.add_nodes(nodes)
|
|
self.assertEqual(count, 2)
|
|
self.assertIn("n1", graph.nodes)
|
|
self.assertIn("n2", graph.nodes)
|
|
|
|
# Add edges
|
|
edges = [
|
|
{"source_id": "n1", "target_id": "n2", "type": "knows", "weight": 0.8}
|
|
]
|
|
count = graph.add_edges(edges)
|
|
self.assertEqual(count, 1)
|
|
self.assertEqual(len(graph.edges), 1)
|
|
|
|
# Get neighbors
|
|
neighbors = graph.get_neighbors("n1")
|
|
self.assertEqual(len(neighbors), 1)
|
|
self.assertEqual(neighbors[0]["id"], "n2")
|
|
self.assertEqual(neighbors[0]["relationship"], "knows")
|
|
|
|
def test_get_nodes_by_label_returns_metadata_copy(self):
|
|
graph = ContextGraph()
|
|
graph.add_node("n1", "person", "Alice", role="engineer")
|
|
|
|
nodes = graph.get_nodes_by_label("person")
|
|
self.assertEqual(len(nodes), 1)
|
|
|
|
nodes[0]["metadata"]["role"] = "mutated"
|
|
|
|
self.assertEqual(graph.get_node_property("n1", "role"), "engineer")
|
|
|
|
def test_context_graph_preserves_full_decision_text(self):
|
|
graph = ContextGraph()
|
|
scenario = "Launch regional expansion plan " + ("X" * 140)
|
|
root_id = graph.record_decision(
|
|
category="strategy",
|
|
scenario=scenario,
|
|
reasoning="Growth opportunity with strong local demand",
|
|
outcome="approved",
|
|
confidence=0.91,
|
|
)
|
|
child_id = graph.record_decision(
|
|
category="operations",
|
|
scenario="Open Sao Paulo office",
|
|
reasoning="Needed to support expansion",
|
|
outcome="pending",
|
|
confidence=0.74,
|
|
)
|
|
graph.add_causal_relationship(root_id, child_id, "CAUSED")
|
|
|
|
chain = graph.get_causal_chain(child_id)
|
|
|
|
self.assertEqual(graph.nodes[root_id].content, scenario)
|
|
self.assertEqual(graph.nodes[root_id].properties["scenario"], scenario)
|
|
self.assertEqual(chain[0].scenario, scenario)
|
|
|
|
def test_record_decision_metadata_cannot_override_core_fields(self):
|
|
graph = ContextGraph()
|
|
decision_id = graph.record_decision(
|
|
category="strategy",
|
|
scenario="Open LATAM expansion program",
|
|
reasoning="High growth potential",
|
|
outcome="approved",
|
|
confidence=0.9,
|
|
metadata={
|
|
"scenario": "metadata override",
|
|
"category": "metadata category",
|
|
"outcome": "metadata outcome",
|
|
"custom_note": "keep me",
|
|
},
|
|
)
|
|
|
|
node = graph.nodes[decision_id]
|
|
|
|
self.assertEqual(node.content, "Open LATAM expansion program")
|
|
self.assertEqual(node.properties["scenario"], "Open LATAM expansion program")
|
|
self.assertEqual(node.properties["category"], "strategy")
|
|
self.assertEqual(node.properties["outcome"], "approved")
|
|
self.assertEqual(node.properties["custom_note"], "keep me")
|
|
|
|
# --- AgentMemory Tests ---
|
|
def test_agent_memory_store(self):
|
|
memory = AgentMemory(vector_store=self.mock_vector_store)
|
|
|
|
# Store item
|
|
memory_id = memory.store("Test memory content", metadata={"type": "test"})
|
|
|
|
self.assertIsNotNone(memory_id)
|
|
self.assertEqual(len(memory.short_term_memory), 1)
|
|
self.assertEqual(memory.short_term_memory[0].content, "Test memory content")
|
|
|
|
# Check vector store interaction (mocked _generate_embedding might be needed if not implemented)
|
|
# The store method calls _generate_embedding. If it's not implemented or relies on external service, it might fail.
|
|
# Let's see if we need to mock _generate_embedding.
|
|
|
|
@patch('semantica.context.agent_memory.AgentMemory._generate_embedding')
|
|
def test_agent_memory_vector_store(self, mock_gen_embedding):
|
|
mock_gen_embedding.return_value = [0.1, 0.2, 0.3]
|
|
memory = AgentMemory(vector_store=self.mock_vector_store)
|
|
|
|
memory.store("Vector test")
|
|
|
|
self.assertEqual(len(self.mock_vector_store.metadata), 1)
|
|
self.assertEqual(self.mock_vector_store.metadata[0].get("type"), None) # Default empty metadata
|
|
|
|
# --- ContextRetriever Tests ---
|
|
def test_context_retriever_init(self):
|
|
retriever = ContextRetriever(
|
|
memory_store=MagicMock(),
|
|
knowledge_graph=MagicMock(),
|
|
vector_store=self.mock_vector_store
|
|
)
|
|
self.assertIsNotNone(retriever)
|
|
|
|
def test_context_graph_rejects_cyclic_skos_single_edge_write(self):
|
|
graph = ContextGraph()
|
|
graph.add_edge("A", "B", "skos:broader")
|
|
|
|
with self.assertRaisesRegex(ValueError, "SKOS hierarchy contains a cycle"):
|
|
graph.add_edge("B", "A", "skos:broader")
|
|
|
|
self.assertEqual(len(graph.edges), 1)
|
|
|
|
def test_context_graph_rejects_cyclic_skos_batch_write(self):
|
|
graph = ContextGraph()
|
|
|
|
with self.assertRaisesRegex(ValueError, "SKOS hierarchy contains a cycle"):
|
|
graph.add_edges([
|
|
{"source": "A", "target": "B", "type": "skos:broader"},
|
|
{"source": "A", "target": "B", "type": "skos:narrower"},
|
|
])
|
|
|
|
self.assertEqual(len(graph.edges), 0)
|
|
|
|
def test_context_graph_preexisting_unrelated_cycle_does_not_block_new_write(self):
|
|
"""A cycle already persisted elsewhere in the graph (e.g. legacy data
|
|
written before cycle detection existed) must not poison unrelated
|
|
SKOS hierarchy writes for concepts it doesn't touch."""
|
|
from semantica.context.context_graph import ContextEdge
|
|
|
|
graph = ContextGraph()
|
|
graph._add_internal_edge(ContextEdge(source_id="X", target_id="Y", edge_type="skos:broader"))
|
|
graph._add_internal_edge(ContextEdge(source_id="Y", target_id="X", edge_type="skos:broader"))
|
|
|
|
self.assertTrue(graph.add_edge("C", "D", "skos:broader"))
|
|
self.assertEqual(len(graph.edges), 3)
|
|
|
|
# --- AgentContext Tests ---
|
|
@patch('semantica.context.agent_memory.AgentMemory._generate_embedding')
|
|
def test_agent_context_end_to_end(self, mock_gen_embedding):
|
|
mock_gen_embedding.return_value = [0.1, 0.1]
|
|
|
|
# Setup complete context system
|
|
kg = ContextGraph()
|
|
ctx = AgentContext(vector_store=self.mock_vector_store, knowledge_graph=kg)
|
|
|
|
# Test store
|
|
ctx.store("Alice knows Bob", extract_entities=False)
|
|
|
|
# Verify internal components
|
|
self.assertIsNotNone(ctx._memory)
|
|
self.assertEqual(len(ctx._memory.short_term_memory), 1)
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|