diff --git a/tests/unit/test_provenance/__init__.py b/tests/unit/test_provenance/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/test_provenance/test_agent_provenance.py b/tests/unit/test_provenance/test_agent_provenance.py new file mode 100644 index 00000000..9377fe19 --- /dev/null +++ b/tests/unit/test_provenance/test_agent_provenance.py @@ -0,0 +1,324 @@ +""" +Tests for agent provenance triple builder functions. +""" + +import json +import pytest + +from trustgraph.schema import Triple, Term, IRI, LITERAL + +from trustgraph.provenance.agent import ( + agent_session_triples, + agent_iteration_triples, + agent_final_triples, +) + +from trustgraph.provenance.namespaces import ( + RDF_TYPE, RDFS_LABEL, + PROV_ACTIVITY, PROV_ENTITY, PROV_WAS_DERIVED_FROM, PROV_STARTED_AT_TIME, + TG_QUERY, TG_THOUGHT, TG_ACTION, TG_ARGUMENTS, TG_OBSERVATION, TG_ANSWER, + TG_QUESTION, TG_ANALYSIS, TG_CONCLUSION, TG_DOCUMENT, + TG_THOUGHT_DOCUMENT, TG_OBSERVATION_DOCUMENT, + TG_AGENT_QUESTION, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def find_triple(triples, predicate, subject=None): + for t in triples: + if t.p.iri == predicate: + if subject is None or t.s.iri == subject: + return t + return None + + +def find_triples(triples, predicate, subject=None): + return [ + t for t in triples + if t.p.iri == predicate and (subject is None or t.s.iri == subject) + ] + + +def has_type(triples, subject, rdf_type): + for t in triples: + if (t.s.iri == subject and t.p.iri == RDF_TYPE + and t.o.type == IRI and t.o.iri == rdf_type): + return True + return False + + +# --------------------------------------------------------------------------- +# agent_session_triples +# --------------------------------------------------------------------------- + +class TestAgentSessionTriples: + + SESSION_URI = "urn:trustgraph:agent:test-session" + + def test_session_types(self): + triples = agent_session_triples( + self.SESSION_URI, "What is X?", "2024-01-01T00:00:00Z" + ) + assert has_type(triples, self.SESSION_URI, PROV_ACTIVITY) + assert has_type(triples, self.SESSION_URI, TG_QUESTION) + assert has_type(triples, self.SESSION_URI, TG_AGENT_QUESTION) + + def test_session_query_text(self): + triples = agent_session_triples( + self.SESSION_URI, "What is X?", "2024-01-01T00:00:00Z" + ) + query = find_triple(triples, TG_QUERY, self.SESSION_URI) + assert query is not None + assert query.o.value == "What is X?" + + def test_session_timestamp(self): + triples = agent_session_triples( + self.SESSION_URI, "Q", "2024-06-15T10:00:00Z" + ) + ts = find_triple(triples, PROV_STARTED_AT_TIME, self.SESSION_URI) + assert ts is not None + assert ts.o.value == "2024-06-15T10:00:00Z" + + def test_session_default_timestamp(self): + triples = agent_session_triples(self.SESSION_URI, "Q") + ts = find_triple(triples, PROV_STARTED_AT_TIME, self.SESSION_URI) + assert ts is not None + assert len(ts.o.value) > 0 + + def test_session_label(self): + triples = agent_session_triples( + self.SESSION_URI, "Q", "2024-01-01T00:00:00Z" + ) + label = find_triple(triples, RDFS_LABEL, self.SESSION_URI) + assert label is not None + assert label.o.value == "Agent Question" + + def test_session_triple_count(self): + triples = agent_session_triples( + self.SESSION_URI, "Q", "2024-01-01T00:00:00Z" + ) + assert len(triples) == 6 + + +# --------------------------------------------------------------------------- +# agent_iteration_triples +# --------------------------------------------------------------------------- + +class TestAgentIterationTriples: + + ITER_URI = "urn:trustgraph:agent:test-session/i1" + PARENT_URI = "urn:trustgraph:agent:test-session" + + def test_iteration_types(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + thought="thinking", action="search", observation="found it", + ) + assert has_type(triples, self.ITER_URI, PROV_ENTITY) + assert has_type(triples, self.ITER_URI, TG_ANALYSIS) + + def test_iteration_derived_from_parent(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="search", + ) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.ITER_URI) + assert derived is not None + assert derived.o.iri == self.PARENT_URI + + def test_iteration_label_includes_action(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="graph-rag-query", + ) + label = find_triple(triples, RDFS_LABEL, self.ITER_URI) + assert label is not None + assert "graph-rag-query" in label.o.value + + def test_iteration_thought_inline(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + thought="I need to search for info", + action="search", + ) + thought = find_triple(triples, TG_THOUGHT, self.ITER_URI) + assert thought is not None + assert thought.o.value == "I need to search for info" + + def test_iteration_thought_document_preferred(self): + """When thought_document_id is provided, inline thought is not stored.""" + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + thought="inline thought", + action="search", + thought_document_id="urn:doc:thought-1", + ) + thought_doc = find_triple(triples, TG_THOUGHT_DOCUMENT, self.ITER_URI) + assert thought_doc is not None + assert thought_doc.o.iri == "urn:doc:thought-1" + thought_inline = find_triple(triples, TG_THOUGHT, self.ITER_URI) + assert thought_inline is None + + def test_iteration_observation_inline(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="search", + observation="Found 3 results", + ) + obs = find_triple(triples, TG_OBSERVATION, self.ITER_URI) + assert obs is not None + assert obs.o.value == "Found 3 results" + + def test_iteration_observation_document_preferred(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="search", + observation="inline obs", + observation_document_id="urn:doc:obs-1", + ) + obs_doc = find_triple(triples, TG_OBSERVATION_DOCUMENT, self.ITER_URI) + assert obs_doc is not None + assert obs_doc.o.iri == "urn:doc:obs-1" + obs_inline = find_triple(triples, TG_OBSERVATION, self.ITER_URI) + assert obs_inline is None + + def test_iteration_action_recorded(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="graph-rag-query", + ) + action = find_triple(triples, TG_ACTION, self.ITER_URI) + assert action is not None + assert action.o.value == "graph-rag-query" + + def test_iteration_arguments_json_encoded(self): + args = {"query": "test query", "limit": 10} + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="search", + arguments=args, + ) + arguments = find_triple(triples, TG_ARGUMENTS, self.ITER_URI) + assert arguments is not None + parsed = json.loads(arguments.o.value) + assert parsed == args + + def test_iteration_default_arguments_empty_dict(self): + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="search", + ) + arguments = find_triple(triples, TG_ARGUMENTS, self.ITER_URI) + assert arguments is not None + parsed = json.loads(arguments.o.value) + assert parsed == {} + + def test_iteration_no_thought_or_observation(self): + """Minimal iteration with just action — no thought or observation triples.""" + triples = agent_iteration_triples( + self.ITER_URI, self.PARENT_URI, + action="noop", + ) + thought = find_triple(triples, TG_THOUGHT, self.ITER_URI) + obs = find_triple(triples, TG_OBSERVATION, self.ITER_URI) + assert thought is None + assert obs is None + + def test_iteration_chaining(self): + """Second iteration derives from first iteration, not session.""" + iter1_uri = "urn:trustgraph:agent:sess/i1" + iter2_uri = "urn:trustgraph:agent:sess/i2" + + triples1 = agent_iteration_triples( + iter1_uri, self.PARENT_URI, action="step1", + ) + triples2 = agent_iteration_triples( + iter2_uri, iter1_uri, action="step2", + ) + + derived1 = find_triple(triples1, PROV_WAS_DERIVED_FROM, iter1_uri) + assert derived1.o.iri == self.PARENT_URI + + derived2 = find_triple(triples2, PROV_WAS_DERIVED_FROM, iter2_uri) + assert derived2.o.iri == iter1_uri + + +# --------------------------------------------------------------------------- +# agent_final_triples +# --------------------------------------------------------------------------- + +class TestAgentFinalTriples: + + FINAL_URI = "urn:trustgraph:agent:test-session/final" + PARENT_URI = "urn:trustgraph:agent:test-session/i3" + + def test_final_types(self): + triples = agent_final_triples( + self.FINAL_URI, self.PARENT_URI, answer="42" + ) + assert has_type(triples, self.FINAL_URI, PROV_ENTITY) + assert has_type(triples, self.FINAL_URI, TG_CONCLUSION) + + def test_final_derived_from_parent(self): + triples = agent_final_triples( + self.FINAL_URI, self.PARENT_URI, answer="42" + ) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.FINAL_URI) + assert derived is not None + assert derived.o.iri == self.PARENT_URI + + def test_final_label(self): + triples = agent_final_triples( + self.FINAL_URI, self.PARENT_URI, answer="42" + ) + label = find_triple(triples, RDFS_LABEL, self.FINAL_URI) + assert label is not None + assert label.o.value == "Conclusion" + + def test_final_inline_answer(self): + triples = agent_final_triples( + self.FINAL_URI, self.PARENT_URI, answer="The answer is 42" + ) + answer = find_triple(triples, TG_ANSWER, self.FINAL_URI) + assert answer is not None + assert answer.o.value == "The answer is 42" + + def test_final_document_reference(self): + triples = agent_final_triples( + self.FINAL_URI, self.PARENT_URI, + document_id="urn:trustgraph:agent:sess/answer", + ) + doc = find_triple(triples, TG_DOCUMENT, self.FINAL_URI) + assert doc is not None + assert doc.o.type == IRI + assert doc.o.iri == "urn:trustgraph:agent:sess/answer" + + def test_final_document_takes_precedence(self): + triples = agent_final_triples( + self.FINAL_URI, self.PARENT_URI, + answer="inline", + document_id="urn:doc:123", + ) + doc = find_triple(triples, TG_DOCUMENT, self.FINAL_URI) + assert doc is not None + answer = find_triple(triples, TG_ANSWER, self.FINAL_URI) + assert answer is None + + def test_final_no_answer_or_document(self): + triples = agent_final_triples(self.FINAL_URI, self.PARENT_URI) + answer = find_triple(triples, TG_ANSWER, self.FINAL_URI) + doc = find_triple(triples, TG_DOCUMENT, self.FINAL_URI) + assert answer is None + assert doc is None + + def test_final_derives_from_session_when_no_iterations(self): + """When agent answers immediately, final derives from session.""" + session_uri = "urn:trustgraph:agent:test-session" + triples = agent_final_triples( + self.FINAL_URI, session_uri, answer="direct answer" + ) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.FINAL_URI) + assert derived.o.iri == session_uri diff --git a/tests/unit/test_provenance/test_explainability.py b/tests/unit/test_provenance/test_explainability.py new file mode 100644 index 00000000..1f27cc61 --- /dev/null +++ b/tests/unit/test_provenance/test_explainability.py @@ -0,0 +1,563 @@ +""" +Tests for the explainability API (entity parsing, wire format conversion, +and ExplainabilityClient). +""" + +import pytest +from unittest.mock import MagicMock, patch + +from trustgraph.api.explainability import ( + EdgeSelection, + ExplainEntity, + Question, + Exploration, + Focus, + Synthesis, + Analysis, + Conclusion, + parse_edge_selection_triples, + extract_term_value, + wire_triples_to_tuples, + ExplainabilityClient, + TG_QUERY, TG_EDGE_COUNT, TG_SELECTED_EDGE, TG_EDGE, TG_REASONING, + TG_CONTENT, TG_DOCUMENT, TG_CHUNK_COUNT, + TG_THOUGHT, TG_ACTION, TG_ARGUMENTS, TG_OBSERVATION, TG_ANSWER, + TG_THOUGHT_DOCUMENT, TG_OBSERVATION_DOCUMENT, + TG_QUESTION, TG_EXPLORATION, TG_FOCUS, TG_SYNTHESIS, + TG_ANALYSIS, TG_CONCLUSION, + TG_GRAPH_RAG_QUESTION, TG_DOC_RAG_QUESTION, TG_AGENT_QUESTION, + PROV_STARTED_AT_TIME, PROV_WAS_DERIVED_FROM, PROV_WAS_GENERATED_BY, + RDF_TYPE, RDFS_LABEL, +) + + +# --------------------------------------------------------------------------- +# Entity from_triples parsing +# --------------------------------------------------------------------------- + +class TestExplainEntityFromTriples: + """Test ExplainEntity.from_triples dispatches to correct subclass.""" + + def test_graphrag_question(self): + triples = [ + ("urn:q:1", RDF_TYPE, TG_QUESTION), + ("urn:q:1", RDF_TYPE, TG_GRAPH_RAG_QUESTION), + ("urn:q:1", TG_QUERY, "What is AI?"), + ("urn:q:1", PROV_STARTED_AT_TIME, "2024-01-01T00:00:00Z"), + ] + entity = ExplainEntity.from_triples("urn:q:1", triples) + assert isinstance(entity, Question) + assert entity.query == "What is AI?" + assert entity.timestamp == "2024-01-01T00:00:00Z" + assert entity.question_type == "graph-rag" + + def test_docrag_question(self): + triples = [ + ("urn:q:2", RDF_TYPE, TG_QUESTION), + ("urn:q:2", RDF_TYPE, TG_DOC_RAG_QUESTION), + ("urn:q:2", TG_QUERY, "Find info"), + ] + entity = ExplainEntity.from_triples("urn:q:2", triples) + assert isinstance(entity, Question) + assert entity.question_type == "document-rag" + + def test_agent_question(self): + triples = [ + ("urn:q:3", RDF_TYPE, TG_QUESTION), + ("urn:q:3", RDF_TYPE, TG_AGENT_QUESTION), + ("urn:q:3", TG_QUERY, "Agent query"), + ] + entity = ExplainEntity.from_triples("urn:q:3", triples) + assert isinstance(entity, Question) + assert entity.question_type == "agent" + + def test_exploration(self): + triples = [ + ("urn:exp:1", RDF_TYPE, TG_EXPLORATION), + ("urn:exp:1", TG_EDGE_COUNT, "15"), + ] + entity = ExplainEntity.from_triples("urn:exp:1", triples) + assert isinstance(entity, Exploration) + assert entity.edge_count == 15 + + def test_exploration_with_chunk_count(self): + triples = [ + ("urn:exp:2", RDF_TYPE, TG_EXPLORATION), + ("urn:exp:2", TG_CHUNK_COUNT, "5"), + ] + entity = ExplainEntity.from_triples("urn:exp:2", triples) + assert isinstance(entity, Exploration) + assert entity.chunk_count == 5 + + def test_exploration_invalid_count(self): + triples = [ + ("urn:exp:3", RDF_TYPE, TG_EXPLORATION), + ("urn:exp:3", TG_EDGE_COUNT, "not-a-number"), + ] + entity = ExplainEntity.from_triples("urn:exp:3", triples) + assert isinstance(entity, Exploration) + assert entity.edge_count == 0 + + def test_focus(self): + triples = [ + ("urn:foc:1", RDF_TYPE, TG_FOCUS), + ("urn:foc:1", TG_SELECTED_EDGE, "urn:edge:1"), + ("urn:foc:1", TG_SELECTED_EDGE, "urn:edge:2"), + ] + entity = ExplainEntity.from_triples("urn:foc:1", triples) + assert isinstance(entity, Focus) + assert len(entity.selected_edge_uris) == 2 + assert "urn:edge:1" in entity.selected_edge_uris + assert "urn:edge:2" in entity.selected_edge_uris + + def test_synthesis_with_content(self): + triples = [ + ("urn:syn:1", RDF_TYPE, TG_SYNTHESIS), + ("urn:syn:1", TG_CONTENT, "The answer is 42"), + ] + entity = ExplainEntity.from_triples("urn:syn:1", triples) + assert isinstance(entity, Synthesis) + assert entity.content == "The answer is 42" + assert entity.document_uri == "" + + def test_synthesis_with_document(self): + triples = [ + ("urn:syn:2", RDF_TYPE, TG_SYNTHESIS), + ("urn:syn:2", TG_DOCUMENT, "urn:doc:answer-1"), + ] + entity = ExplainEntity.from_triples("urn:syn:2", triples) + assert isinstance(entity, Synthesis) + assert entity.document_uri == "urn:doc:answer-1" + + def test_analysis(self): + triples = [ + ("urn:ana:1", RDF_TYPE, TG_ANALYSIS), + ("urn:ana:1", TG_THOUGHT, "I should search"), + ("urn:ana:1", TG_ACTION, "graph-rag-query"), + ("urn:ana:1", TG_ARGUMENTS, '{"query": "test"}'), + ("urn:ana:1", TG_OBSERVATION, "Found results"), + ] + entity = ExplainEntity.from_triples("urn:ana:1", triples) + assert isinstance(entity, Analysis) + assert entity.thought == "I should search" + assert entity.action == "graph-rag-query" + assert entity.arguments == '{"query": "test"}' + assert entity.observation == "Found results" + + def test_analysis_with_document_refs(self): + triples = [ + ("urn:ana:2", RDF_TYPE, TG_ANALYSIS), + ("urn:ana:2", TG_ACTION, "search"), + ("urn:ana:2", TG_THOUGHT_DOCUMENT, "urn:doc:thought-1"), + ("urn:ana:2", TG_OBSERVATION_DOCUMENT, "urn:doc:obs-1"), + ] + entity = ExplainEntity.from_triples("urn:ana:2", triples) + assert isinstance(entity, Analysis) + assert entity.thought_document_uri == "urn:doc:thought-1" + assert entity.observation_document_uri == "urn:doc:obs-1" + + def test_conclusion_with_answer(self): + triples = [ + ("urn:conc:1", RDF_TYPE, TG_CONCLUSION), + ("urn:conc:1", TG_ANSWER, "The final answer"), + ] + entity = ExplainEntity.from_triples("urn:conc:1", triples) + assert isinstance(entity, Conclusion) + assert entity.answer == "The final answer" + + def test_conclusion_with_document(self): + triples = [ + ("urn:conc:2", RDF_TYPE, TG_CONCLUSION), + ("urn:conc:2", TG_DOCUMENT, "urn:doc:final"), + ] + entity = ExplainEntity.from_triples("urn:conc:2", triples) + assert isinstance(entity, Conclusion) + assert entity.document_uri == "urn:doc:final" + + def test_unknown_type(self): + triples = [ + ("urn:x:1", RDF_TYPE, "http://example.com/UnknownType"), + ] + entity = ExplainEntity.from_triples("urn:x:1", triples) + assert isinstance(entity, ExplainEntity) + assert entity.entity_type == "unknown" + + +# --------------------------------------------------------------------------- +# parse_edge_selection_triples +# --------------------------------------------------------------------------- + +class TestParseEdgeSelectionTriples: + + def test_with_edge_and_reasoning(self): + triples = [ + ("urn:edge:1", TG_EDGE, {"s": "Alice", "p": "knows", "o": "Bob"}), + ("urn:edge:1", TG_REASONING, "Alice and Bob are connected"), + ] + result = parse_edge_selection_triples(triples) + assert isinstance(result, EdgeSelection) + assert result.uri == "urn:edge:1" + assert result.edge == {"s": "Alice", "p": "knows", "o": "Bob"} + assert result.reasoning == "Alice and Bob are connected" + + def test_with_edge_only(self): + triples = [ + ("urn:edge:2", TG_EDGE, {"s": "A", "p": "r", "o": "B"}), + ] + result = parse_edge_selection_triples(triples) + assert result.edge is not None + assert result.reasoning == "" + + def test_with_reasoning_only(self): + triples = [ + ("urn:edge:3", TG_REASONING, "some reason"), + ] + result = parse_edge_selection_triples(triples) + assert result.edge is None + assert result.reasoning == "some reason" + + def test_empty_triples(self): + result = parse_edge_selection_triples([]) + assert result.uri == "" + assert result.edge is None + assert result.reasoning == "" + + def test_edge_must_be_dict(self): + """Non-dict values for TG_EDGE should not be treated as edges.""" + triples = [ + ("urn:edge:4", TG_EDGE, "not-a-dict"), + ] + result = parse_edge_selection_triples(triples) + assert result.edge is None + + +# --------------------------------------------------------------------------- +# extract_term_value +# --------------------------------------------------------------------------- + +class TestExtractTermValue: + + def test_iri_short_format(self): + assert extract_term_value({"t": "i", "i": "urn:test"}) == "urn:test" + + def test_iri_long_format(self): + assert extract_term_value({"type": "i", "iri": "urn:test"}) == "urn:test" + + def test_literal_short_format(self): + assert extract_term_value({"t": "l", "v": "hello"}) == "hello" + + def test_literal_long_format(self): + assert extract_term_value({"type": "l", "value": "hello"}) == "hello" + + def test_quoted_triple(self): + term = { + "t": "t", + "tr": { + "s": {"t": "i", "i": "urn:s"}, + "p": {"t": "i", "i": "urn:p"}, + "o": {"t": "i", "i": "urn:o"}, + } + } + result = extract_term_value(term) + assert result == {"s": "urn:s", "p": "urn:p", "o": "urn:o"} + + def test_quoted_triple_long_format(self): + term = { + "type": "t", + "triple": { + "s": {"type": "i", "iri": "urn:s"}, + "p": {"type": "i", "iri": "urn:p"}, + "o": {"type": "l", "value": "val"}, + } + } + result = extract_term_value(term) + assert result == {"s": "urn:s", "p": "urn:p", "o": "val"} + + def test_unknown_type_fallback(self): + result = extract_term_value({"t": "x", "i": "urn:fallback"}) + assert result == "urn:fallback" + + +# --------------------------------------------------------------------------- +# wire_triples_to_tuples +# --------------------------------------------------------------------------- + +class TestWireTriplesToTuples: + + def test_basic_conversion(self): + wire = [ + { + "s": {"t": "i", "i": "urn:s1"}, + "p": {"t": "i", "i": "urn:p1"}, + "o": {"t": "l", "v": "value1"}, + }, + ] + result = wire_triples_to_tuples(wire) + assert len(result) == 1 + assert result[0] == ("urn:s1", "urn:p1", "value1") + + def test_multiple_triples(self): + wire = [ + { + "s": {"t": "i", "i": "urn:s1"}, + "p": {"t": "i", "i": "urn:p1"}, + "o": {"t": "l", "v": "v1"}, + }, + { + "s": {"t": "i", "i": "urn:s2"}, + "p": {"t": "i", "i": "urn:p2"}, + "o": {"t": "i", "i": "urn:o2"}, + }, + ] + result = wire_triples_to_tuples(wire) + assert len(result) == 2 + assert result[0] == ("urn:s1", "urn:p1", "v1") + assert result[1] == ("urn:s2", "urn:p2", "urn:o2") + + def test_empty_list(self): + assert wire_triples_to_tuples([]) == [] + + def test_missing_fields(self): + wire = [{"s": {}, "p": {}, "o": {}}] + result = wire_triples_to_tuples(wire) + assert len(result) == 1 + + +# --------------------------------------------------------------------------- +# ExplainabilityClient +# --------------------------------------------------------------------------- + +def _make_wire_triples(tuples): + """Convert (s, p, o) tuples to wire format for mocking.""" + result = [] + for s, p, o in tuples: + entry = { + "s": {"t": "i", "i": s}, + "p": {"t": "i", "i": p}, + } + if o.startswith("urn:") or o.startswith("http"): + entry["o"] = {"t": "i", "i": o} + else: + entry["o"] = {"t": "l", "v": o} + result.append(entry) + return result + + +class TestExplainabilityClientFetchEntity: + + def test_fetch_question_entity(self): + wire = _make_wire_triples([ + ("urn:q:1", RDF_TYPE, TG_QUESTION), + ("urn:q:1", RDF_TYPE, TG_GRAPH_RAG_QUESTION), + ("urn:q:1", TG_QUERY, "What is AI?"), + ("urn:q:1", PROV_STARTED_AT_TIME, "2024-01-01T00:00:00Z"), + ]) + + mock_flow = MagicMock() + # Return same results twice for quiescence + mock_flow.triples_query.side_effect = [wire, wire] + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + entity = client.fetch_entity("urn:q:1", graph="urn:graph:retrieval") + + assert isinstance(entity, Question) + assert entity.query == "What is AI?" + assert entity.question_type == "graph-rag" + + def test_fetch_returns_none_when_no_data(self): + mock_flow = MagicMock() + mock_flow.triples_query.return_value = [] + + client = ExplainabilityClient(mock_flow, retry_delay=0.0, max_retries=2) + entity = client.fetch_entity("urn:nonexistent") + + assert entity is None + + def test_fetch_retries_on_empty_results(self): + wire = _make_wire_triples([ + ("urn:q:1", RDF_TYPE, TG_QUESTION), + ("urn:q:1", RDF_TYPE, TG_GRAPH_RAG_QUESTION), + ("urn:q:1", TG_QUERY, "Q"), + ]) + + mock_flow = MagicMock() + # Empty, then data, then same data (stable) + mock_flow.triples_query.side_effect = [[], wire, wire] + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + entity = client.fetch_entity("urn:q:1") + + assert isinstance(entity, Question) + assert mock_flow.triples_query.call_count == 3 + + +class TestExplainabilityClientResolveLabel: + + def test_resolve_label_found(self): + mock_flow = MagicMock() + mock_flow.triples_query.return_value = _make_wire_triples([ + ("urn:entity:1", RDFS_LABEL, "Entity One"), + ]) + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + label = client.resolve_label("urn:entity:1") + assert label == "Entity One" + + def test_resolve_label_not_found(self): + mock_flow = MagicMock() + mock_flow.triples_query.return_value = [] + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + label = client.resolve_label("urn:entity:1") + assert label == "urn:entity:1" + + def test_resolve_label_cached(self): + mock_flow = MagicMock() + mock_flow.triples_query.return_value = _make_wire_triples([ + ("urn:entity:1", RDFS_LABEL, "Entity One"), + ]) + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + client.resolve_label("urn:entity:1") + client.resolve_label("urn:entity:1") + + # Only one query should be made + assert mock_flow.triples_query.call_count == 1 + + def test_resolve_label_non_uri(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + assert client.resolve_label("plain text") == "plain text" + assert client.resolve_label("") == "" + mock_flow.triples_query.assert_not_called() + + def test_resolve_edge_labels(self): + mock_flow = MagicMock() + + def mock_query(s=None, p=None, **kwargs): + labels = { + "urn:e:Alice": "Alice", + "urn:r:knows": "knows", + "urn:e:Bob": "Bob", + } + if s in labels: + return _make_wire_triples([(s, RDFS_LABEL, labels[s])]) + return [] + + mock_flow.triples_query.side_effect = mock_query + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + s, p, o = client.resolve_edge_labels( + {"s": "urn:e:Alice", "p": "urn:r:knows", "o": "urn:e:Bob"} + ) + assert s == "Alice" + assert p == "knows" + assert o == "Bob" + + +class TestExplainabilityClientContentFetching: + + def test_fetch_synthesis_inline_content(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + + synthesis = Synthesis(uri="urn:syn:1", content="inline answer") + result = client.fetch_synthesis_content(synthesis, api=None) + assert result == "inline answer" + + def test_fetch_synthesis_truncated_content(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + + long_content = "x" * 20000 + synthesis = Synthesis(uri="urn:syn:1", content=long_content) + result = client.fetch_synthesis_content(synthesis, api=None, max_content=100) + assert len(result) < 20000 + assert result.endswith("... [truncated]") + + def test_fetch_synthesis_from_librarian(self): + mock_flow = MagicMock() + mock_api = MagicMock() + mock_library = MagicMock() + mock_api.library.return_value = mock_library + mock_library.get_document_content.return_value = b"librarian content" + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + synthesis = Synthesis( + uri="urn:syn:1", + document_uri="urn:document:abc123" + ) + result = client.fetch_synthesis_content(synthesis, api=mock_api) + assert result == "librarian content" + + def test_fetch_synthesis_no_content_or_document(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + + synthesis = Synthesis(uri="urn:syn:1") + result = client.fetch_synthesis_content(synthesis, api=None) + assert result == "" + + def test_fetch_conclusion_inline(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + + conclusion = Conclusion(uri="urn:conc:1", answer="42") + result = client.fetch_conclusion_content(conclusion, api=None) + assert result == "42" + + def test_fetch_analysis_content_from_librarian(self): + mock_flow = MagicMock() + mock_api = MagicMock() + mock_library = MagicMock() + mock_api.library.return_value = mock_library + mock_library.get_document_content.side_effect = [ + b"thought content", + b"observation content", + ] + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + analysis = Analysis( + uri="urn:ana:1", + action="search", + thought_document_uri="urn:doc:thought", + observation_document_uri="urn:doc:obs", + ) + client.fetch_analysis_content(analysis, api=mock_api) + assert analysis.thought == "thought content" + assert analysis.observation == "observation content" + + def test_fetch_analysis_skips_when_inline_exists(self): + mock_flow = MagicMock() + mock_api = MagicMock() + + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + analysis = Analysis( + uri="urn:ana:1", + action="search", + thought="already have thought", + observation="already have observation", + thought_document_uri="urn:doc:thought", + observation_document_uri="urn:doc:obs", + ) + client.fetch_analysis_content(analysis, api=mock_api) + # Should not call librarian since inline content exists + mock_api.library.assert_not_called() + + +class TestExplainabilityClientDetectSessionType: + + def test_detect_agent_from_uri(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + assert client.detect_session_type("urn:trustgraph:agent:abc") == "agent" + + def test_detect_graphrag_from_uri(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + assert client.detect_session_type("urn:trustgraph:question:abc") == "graphrag" + + def test_detect_docrag_from_uri(self): + mock_flow = MagicMock() + client = ExplainabilityClient(mock_flow, retry_delay=0.0) + assert client.detect_session_type("urn:trustgraph:docrag:abc") == "docrag" diff --git a/tests/unit/test_provenance/test_triples.py b/tests/unit/test_provenance/test_triples.py new file mode 100644 index 00000000..91074097 --- /dev/null +++ b/tests/unit/test_provenance/test_triples.py @@ -0,0 +1,785 @@ +""" +Tests for provenance triple builder functions (extraction-time and query-time). +""" + +import pytest +from unittest.mock import patch + +from trustgraph.schema import Triple, Term, IRI, LITERAL, TRIPLE + +from trustgraph.provenance.triples import ( + set_graph, + document_triples, + derived_entity_triples, + subgraph_provenance_triples, + question_triples, + exploration_triples, + focus_triples, + synthesis_triples, + docrag_question_triples, + docrag_exploration_triples, + docrag_synthesis_triples, +) + +from trustgraph.provenance.namespaces import ( + RDF_TYPE, RDFS_LABEL, + PROV_ENTITY, PROV_ACTIVITY, PROV_AGENT, + PROV_WAS_DERIVED_FROM, PROV_WAS_GENERATED_BY, + PROV_USED, PROV_WAS_ASSOCIATED_WITH, PROV_STARTED_AT_TIME, + DC_TITLE, DC_SOURCE, DC_DATE, DC_CREATOR, + TG_PAGE_COUNT, TG_MIME_TYPE, TG_PAGE_NUMBER, + TG_CHUNK_INDEX, TG_CHAR_OFFSET, TG_CHAR_LENGTH, + TG_CHUNK_SIZE, TG_CHUNK_OVERLAP, TG_COMPONENT_VERSION, + TG_LLM_MODEL, TG_ONTOLOGY, TG_CONTAINS, + TG_DOCUMENT_TYPE, TG_PAGE_TYPE, TG_CHUNK_TYPE, TG_SUBGRAPH_TYPE, + TG_QUERY, TG_EDGE_COUNT, TG_SELECTED_EDGE, TG_EDGE, TG_REASONING, + TG_CONTENT, TG_DOCUMENT, + TG_CHUNK_COUNT, TG_SELECTED_CHUNK, + TG_QUESTION, TG_EXPLORATION, TG_FOCUS, TG_SYNTHESIS, + TG_GRAPH_RAG_QUESTION, TG_DOC_RAG_QUESTION, + GRAPH_SOURCE, GRAPH_RETRIEVAL, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def find_triple(triples, predicate, subject=None): + """Find first triple matching predicate (and optionally subject).""" + for t in triples: + if t.p.iri == predicate: + if subject is None or t.s.iri == subject: + return t + return None + + +def find_triples(triples, predicate, subject=None): + """Find all triples matching predicate (and optionally subject).""" + return [ + t for t in triples + if t.p.iri == predicate and (subject is None or t.s.iri == subject) + ] + + +def has_type(triples, subject, rdf_type): + """Check if subject has rdf:type rdf_type.""" + for t in triples: + if (t.s.iri == subject and t.p.iri == RDF_TYPE + and t.o.type == IRI and t.o.iri == rdf_type): + return True + return False + + +# --------------------------------------------------------------------------- +# set_graph +# --------------------------------------------------------------------------- + +class TestSetGraph: + + def test_sets_graph_on_all_triples(self): + triples = [ + Triple( + s=Term(type=IRI, iri="urn:s1"), + p=Term(type=IRI, iri="urn:p1"), + o=Term(type=LITERAL, value="v1"), + ), + Triple( + s=Term(type=IRI, iri="urn:s2"), + p=Term(type=IRI, iri="urn:p2"), + o=Term(type=LITERAL, value="v2"), + ), + ] + result = set_graph(triples, GRAPH_RETRIEVAL) + assert len(result) == 2 + for t in result: + assert t.g == GRAPH_RETRIEVAL + + def test_does_not_modify_originals(self): + original = Triple( + s=Term(type=IRI, iri="urn:s"), + p=Term(type=IRI, iri="urn:p"), + o=Term(type=LITERAL, value="v"), + ) + result = set_graph([original], "urn:graph:test") + assert original.g is None + assert result[0].g == "urn:graph:test" + + def test_empty_list(self): + result = set_graph([], GRAPH_SOURCE) + assert result == [] + + def test_preserves_spo(self): + original = Triple( + s=Term(type=IRI, iri="urn:s"), + p=Term(type=IRI, iri="urn:p"), + o=Term(type=LITERAL, value="hello"), + ) + result = set_graph([original], "urn:g")[0] + assert result.s.iri == "urn:s" + assert result.p.iri == "urn:p" + assert result.o.value == "hello" + + +# --------------------------------------------------------------------------- +# document_triples +# --------------------------------------------------------------------------- + +class TestDocumentTriples: + + DOC_URI = "https://example.com/doc/abc" + + def test_minimal_document(self): + triples = document_triples(self.DOC_URI) + assert has_type(triples, self.DOC_URI, PROV_ENTITY) + assert has_type(triples, self.DOC_URI, TG_DOCUMENT_TYPE) + assert len(triples) == 2 + + def test_with_title(self): + triples = document_triples(self.DOC_URI, title="My Doc") + title_t = find_triple(triples, DC_TITLE) + assert title_t is not None + assert title_t.o.value == "My Doc" + # Title also creates an rdfs:label + label_t = find_triple(triples, RDFS_LABEL) + assert label_t is not None + assert label_t.o.value == "My Doc" + + def test_with_source(self): + triples = document_triples(self.DOC_URI, source="https://source.com/f.pdf") + source_t = find_triple(triples, DC_SOURCE) + assert source_t is not None + assert source_t.o.type == IRI + assert source_t.o.iri == "https://source.com/f.pdf" + + def test_with_date(self): + triples = document_triples(self.DOC_URI, date="2024-01-15") + date_t = find_triple(triples, DC_DATE) + assert date_t is not None + assert date_t.o.value == "2024-01-15" + + def test_with_creator(self): + triples = document_triples(self.DOC_URI, creator="Alice") + creator_t = find_triple(triples, DC_CREATOR) + assert creator_t is not None + assert creator_t.o.value == "Alice" + + def test_with_page_count(self): + triples = document_triples(self.DOC_URI, page_count=42) + pc_t = find_triple(triples, TG_PAGE_COUNT) + assert pc_t is not None + assert pc_t.o.value == "42" + + def test_with_page_count_zero(self): + triples = document_triples(self.DOC_URI, page_count=0) + pc_t = find_triple(triples, TG_PAGE_COUNT) + assert pc_t is not None + assert pc_t.o.value == "0" + + def test_with_mime_type(self): + triples = document_triples(self.DOC_URI, mime_type="application/pdf") + mt_t = find_triple(triples, TG_MIME_TYPE) + assert mt_t is not None + assert mt_t.o.value == "application/pdf" + + def test_all_metadata(self): + triples = document_triples( + self.DOC_URI, + title="Test", + source="https://s.com", + date="2024-01-01", + creator="Bob", + page_count=10, + mime_type="application/pdf", + ) + # 2 type triples + title + label + source + date + creator + page_count + mime_type + assert len(triples) == 9 + + def test_subject_is_doc_uri(self): + triples = document_triples(self.DOC_URI, title="T") + for t in triples: + assert t.s.iri == self.DOC_URI + + +# --------------------------------------------------------------------------- +# derived_entity_triples +# --------------------------------------------------------------------------- + +class TestDerivedEntityTriples: + + ENTITY_URI = "https://example.com/doc/abc/p1" + PARENT_URI = "https://example.com/doc/abc" + + def test_page_entity_has_page_type(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + page_number=1, + timestamp="2024-01-01T00:00:00Z", + ) + assert has_type(triples, self.ENTITY_URI, PROV_ENTITY) + assert has_type(triples, self.ENTITY_URI, TG_PAGE_TYPE) + + def test_chunk_entity_has_chunk_type(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "chunker", "1.0", + chunk_index=0, + timestamp="2024-01-01T00:00:00Z", + ) + assert has_type(triples, self.ENTITY_URI, TG_CHUNK_TYPE) + + def test_no_specific_type_without_page_or_chunk(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "component", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + assert has_type(triples, self.ENTITY_URI, PROV_ENTITY) + assert not has_type(triples, self.ENTITY_URI, TG_PAGE_TYPE) + assert not has_type(triples, self.ENTITY_URI, TG_CHUNK_TYPE) + + def test_was_derived_from_parent(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.ENTITY_URI) + assert derived is not None + assert derived.o.iri == self.PARENT_URI + + def test_activity_created(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + # Entity was generated by an activity + gen = find_triple(triples, PROV_WAS_GENERATED_BY, self.ENTITY_URI) + assert gen is not None + act_uri = gen.o.iri + + # Activity has correct type and metadata + assert has_type(triples, act_uri, PROV_ACTIVITY) + + # Activity used the parent + used = find_triple(triples, PROV_USED, act_uri) + assert used is not None + assert used.o.iri == self.PARENT_URI + + # Activity has component version + version = find_triple(triples, TG_COMPONENT_VERSION, act_uri) + assert version is not None + assert version.o.value == "1.0" + + def test_agent_created(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + # Find the agent URI via wasAssociatedWith + gen = find_triple(triples, PROV_WAS_GENERATED_BY, self.ENTITY_URI) + act_uri = gen.o.iri + assoc = find_triple(triples, PROV_WAS_ASSOCIATED_WITH, act_uri) + assert assoc is not None + agt_uri = assoc.o.iri + + assert has_type(triples, agt_uri, PROV_AGENT) + label = find_triple(triples, RDFS_LABEL, agt_uri) + assert label is not None + assert label.o.value == "pdf-extractor" + + def test_timestamp_recorded(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + timestamp="2024-06-15T12:30:00Z", + ) + ts = find_triple(triples, PROV_STARTED_AT_TIME) + assert ts is not None + assert ts.o.value == "2024-06-15T12:30:00Z" + + def test_default_timestamp_generated(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + ) + ts = find_triple(triples, PROV_STARTED_AT_TIME) + assert ts is not None + assert len(ts.o.value) > 0 + + def test_optional_label(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + label="Page 1", + timestamp="2024-01-01T00:00:00Z", + ) + label = find_triple(triples, RDFS_LABEL, self.ENTITY_URI) + assert label is not None + assert label.o.value == "Page 1" + + def test_page_number_recorded(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "pdf-extractor", "1.0", + page_number=3, + timestamp="2024-01-01T00:00:00Z", + ) + pn = find_triple(triples, TG_PAGE_NUMBER, self.ENTITY_URI) + assert pn is not None + assert pn.o.value == "3" + + def test_chunk_metadata_recorded(self): + triples = derived_entity_triples( + self.ENTITY_URI, self.PARENT_URI, + "chunker", "2.0", + chunk_index=5, + char_offset=1000, + char_length=500, + chunk_size=512, + chunk_overlap=64, + timestamp="2024-01-01T00:00:00Z", + ) + ci = find_triple(triples, TG_CHUNK_INDEX, self.ENTITY_URI) + assert ci is not None and ci.o.value == "5" + + co = find_triple(triples, TG_CHAR_OFFSET, self.ENTITY_URI) + assert co is not None and co.o.value == "1000" + + cl = find_triple(triples, TG_CHAR_LENGTH, self.ENTITY_URI) + assert cl is not None and cl.o.value == "500" + + # chunk_size and chunk_overlap are on the activity, not the entity + cs = find_triple(triples, TG_CHUNK_SIZE) + assert cs is not None and cs.o.value == "512" + + ov = find_triple(triples, TG_CHUNK_OVERLAP) + assert ov is not None and ov.o.value == "64" + + +# --------------------------------------------------------------------------- +# subgraph_provenance_triples +# --------------------------------------------------------------------------- + +class TestSubgraphProvenanceTriples: + + SG_URI = "https://trustgraph.ai/subgraph/test-sg" + CHUNK_URI = "https://example.com/doc/abc/p1/c0" + + def _make_extracted_triple(self, s="urn:e:Alice", p="urn:r:knows", o="urn:e:Bob"): + return Triple( + s=Term(type=IRI, iri=s), + p=Term(type=IRI, iri=p), + o=Term(type=IRI, iri=o), + ) + + def test_contains_quoted_triples(self): + extracted = [self._make_extracted_triple()] + triples = subgraph_provenance_triples( + self.SG_URI, extracted, self.CHUNK_URI, + "kg-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + contains = find_triples(triples, TG_CONTAINS, self.SG_URI) + assert len(contains) == 1 + assert contains[0].o.type == TRIPLE + assert contains[0].o.triple.s.iri == "urn:e:Alice" + assert contains[0].o.triple.p.iri == "urn:r:knows" + assert contains[0].o.triple.o.iri == "urn:e:Bob" + + def test_multiple_extracted_triples(self): + extracted = [ + self._make_extracted_triple("urn:e:A", "urn:r:x", "urn:e:B"), + self._make_extracted_triple("urn:e:C", "urn:r:y", "urn:e:D"), + self._make_extracted_triple("urn:e:E", "urn:r:z", "urn:e:F"), + ] + triples = subgraph_provenance_triples( + self.SG_URI, extracted, self.CHUNK_URI, + "kg-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + contains = find_triples(triples, TG_CONTAINS, self.SG_URI) + assert len(contains) == 3 + + def test_empty_extracted_triples(self): + triples = subgraph_provenance_triples( + self.SG_URI, [], self.CHUNK_URI, + "kg-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + contains = find_triples(triples, TG_CONTAINS, self.SG_URI) + assert len(contains) == 0 + # Should still have subgraph provenance metadata + assert has_type(triples, self.SG_URI, TG_SUBGRAPH_TYPE) + + def test_subgraph_has_correct_types(self): + triples = subgraph_provenance_triples( + self.SG_URI, [], self.CHUNK_URI, + "kg-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + assert has_type(triples, self.SG_URI, PROV_ENTITY) + assert has_type(triples, self.SG_URI, TG_SUBGRAPH_TYPE) + + def test_derived_from_chunk(self): + triples = subgraph_provenance_triples( + self.SG_URI, [], self.CHUNK_URI, + "kg-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.SG_URI) + assert derived is not None + assert derived.o.iri == self.CHUNK_URI + + def test_activity_and_agent(self): + triples = subgraph_provenance_triples( + self.SG_URI, [], self.CHUNK_URI, + "kg-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + gen = find_triple(triples, PROV_WAS_GENERATED_BY, self.SG_URI) + assert gen is not None + act_uri = gen.o.iri + + assert has_type(triples, act_uri, PROV_ACTIVITY) + + used = find_triple(triples, PROV_USED, act_uri) + assert used is not None + assert used.o.iri == self.CHUNK_URI + + version = find_triple(triples, TG_COMPONENT_VERSION, act_uri) + assert version is not None + assert version.o.value == "1.0" + + def test_optional_llm_model(self): + triples = subgraph_provenance_triples( + self.SG_URI, [], self.CHUNK_URI, + "kg-extractor", "1.0", + llm_model="claude-3-opus", + timestamp="2024-01-01T00:00:00Z", + ) + llm = find_triple(triples, TG_LLM_MODEL) + assert llm is not None + assert llm.o.value == "claude-3-opus" + + def test_no_llm_model_when_omitted(self): + triples = subgraph_provenance_triples( + self.SG_URI, [], self.CHUNK_URI, + "kg-extractor", "1.0", + timestamp="2024-01-01T00:00:00Z", + ) + llm = find_triple(triples, TG_LLM_MODEL) + assert llm is None + + def test_optional_ontology(self): + triples = subgraph_provenance_triples( + self.SG_URI, [], self.CHUNK_URI, + "kg-extractor", "1.0", + ontology_uri="https://example.com/ontology/v1", + timestamp="2024-01-01T00:00:00Z", + ) + ont = find_triple(triples, TG_ONTOLOGY) + assert ont is not None + assert ont.o.type == IRI + assert ont.o.iri == "https://example.com/ontology/v1" + + +# --------------------------------------------------------------------------- +# GraphRAG query-time triples +# --------------------------------------------------------------------------- + +class TestQuestionTriples: + + Q_URI = "urn:trustgraph:question:test-session" + + def test_question_types(self): + triples = question_triples(self.Q_URI, "What is AI?", "2024-01-01T00:00:00Z") + assert has_type(triples, self.Q_URI, PROV_ACTIVITY) + assert has_type(triples, self.Q_URI, TG_QUESTION) + assert has_type(triples, self.Q_URI, TG_GRAPH_RAG_QUESTION) + + def test_question_query_text(self): + triples = question_triples(self.Q_URI, "What is AI?", "2024-01-01T00:00:00Z") + query = find_triple(triples, TG_QUERY, self.Q_URI) + assert query is not None + assert query.o.value == "What is AI?" + + def test_question_timestamp(self): + triples = question_triples(self.Q_URI, "Q", "2024-06-15T10:00:00Z") + ts = find_triple(triples, PROV_STARTED_AT_TIME, self.Q_URI) + assert ts is not None + assert ts.o.value == "2024-06-15T10:00:00Z" + + def test_question_default_timestamp(self): + triples = question_triples(self.Q_URI, "Q") + ts = find_triple(triples, PROV_STARTED_AT_TIME, self.Q_URI) + assert ts is not None + assert len(ts.o.value) > 0 + + def test_question_label(self): + triples = question_triples(self.Q_URI, "Q", "2024-01-01T00:00:00Z") + label = find_triple(triples, RDFS_LABEL, self.Q_URI) + assert label is not None + assert label.o.value == "GraphRAG Question" + + def test_question_triple_count(self): + triples = question_triples(self.Q_URI, "Q", "2024-01-01T00:00:00Z") + assert len(triples) == 6 + + +class TestExplorationTriples: + + EXP_URI = "urn:trustgraph:prov:exploration:test-session" + Q_URI = "urn:trustgraph:question:test-session" + + def test_exploration_types(self): + triples = exploration_triples(self.EXP_URI, self.Q_URI, 15) + assert has_type(triples, self.EXP_URI, PROV_ENTITY) + assert has_type(triples, self.EXP_URI, TG_EXPLORATION) + + def test_exploration_generated_by_question(self): + triples = exploration_triples(self.EXP_URI, self.Q_URI, 15) + gen = find_triple(triples, PROV_WAS_GENERATED_BY, self.EXP_URI) + assert gen is not None + assert gen.o.iri == self.Q_URI + + def test_exploration_edge_count(self): + triples = exploration_triples(self.EXP_URI, self.Q_URI, 15) + ec = find_triple(triples, TG_EDGE_COUNT, self.EXP_URI) + assert ec is not None + assert ec.o.value == "15" + + def test_exploration_zero_edges(self): + triples = exploration_triples(self.EXP_URI, self.Q_URI, 0) + ec = find_triple(triples, TG_EDGE_COUNT, self.EXP_URI) + assert ec is not None + assert ec.o.value == "0" + + def test_exploration_triple_count(self): + triples = exploration_triples(self.EXP_URI, self.Q_URI, 10) + assert len(triples) == 5 + + +class TestFocusTriples: + + FOC_URI = "urn:trustgraph:prov:focus:test-session" + EXP_URI = "urn:trustgraph:prov:exploration:test-session" + SESSION_ID = "test-session" + + def test_focus_types(self): + triples = focus_triples(self.FOC_URI, self.EXP_URI, [], self.SESSION_ID) + assert has_type(triples, self.FOC_URI, PROV_ENTITY) + assert has_type(triples, self.FOC_URI, TG_FOCUS) + + def test_focus_derived_from_exploration(self): + triples = focus_triples(self.FOC_URI, self.EXP_URI, [], self.SESSION_ID) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.FOC_URI) + assert derived is not None + assert derived.o.iri == self.EXP_URI + + def test_focus_no_edges(self): + triples = focus_triples(self.FOC_URI, self.EXP_URI, [], self.SESSION_ID) + selected = find_triples(triples, TG_SELECTED_EDGE) + assert len(selected) == 0 + + def test_focus_with_edges_and_reasoning(self): + edges = [ + { + "edge": ("urn:e:Alice", "urn:r:knows", "urn:e:Bob"), + "reasoning": "Alice is connected to Bob", + }, + { + "edge": ("urn:e:Bob", "urn:r:worksAt", "urn:e:Acme"), + "reasoning": "Bob works at Acme", + }, + ] + triples = focus_triples(self.FOC_URI, self.EXP_URI, edges, self.SESSION_ID) + + # Two selectedEdge links + selected = find_triples(triples, TG_SELECTED_EDGE, self.FOC_URI) + assert len(selected) == 2 + + # Each edge selection has a quoted triple + edge_triples = find_triples(triples, TG_EDGE) + assert len(edge_triples) == 2 + for et in edge_triples: + assert et.o.type == TRIPLE + + # Each edge selection has reasoning + reasoning_triples = find_triples(triples, TG_REASONING) + assert len(reasoning_triples) == 2 + + def test_focus_edge_without_reasoning(self): + edges = [ + {"edge": ("urn:e:A", "urn:r:x", "urn:e:B"), "reasoning": ""}, + ] + triples = focus_triples(self.FOC_URI, self.EXP_URI, edges, self.SESSION_ID) + reasoning = find_triples(triples, TG_REASONING) + assert len(reasoning) == 0 + + def test_focus_edge_without_edge_data(self): + edges = [ + {"edge": None, "reasoning": "some reasoning"}, + ] + triples = focus_triples(self.FOC_URI, self.EXP_URI, edges, self.SESSION_ID) + selected = find_triples(triples, TG_SELECTED_EDGE) + assert len(selected) == 0 + + def test_focus_quoted_triple_content(self): + edges = [ + { + "edge": ("urn:e:Alice", "urn:r:knows", "urn:e:Bob"), + "reasoning": "test", + }, + ] + triples = focus_triples(self.FOC_URI, self.EXP_URI, edges, self.SESSION_ID) + edge_t = find_triple(triples, TG_EDGE) + qt = edge_t.o.triple + assert qt.s.iri == "urn:e:Alice" + assert qt.p.iri == "urn:r:knows" + assert qt.o.iri == "urn:e:Bob" + + +class TestSynthesisTriples: + + SYN_URI = "urn:trustgraph:prov:synthesis:test-session" + FOC_URI = "urn:trustgraph:prov:focus:test-session" + + def test_synthesis_types(self): + triples = synthesis_triples(self.SYN_URI, self.FOC_URI) + assert has_type(triples, self.SYN_URI, PROV_ENTITY) + assert has_type(triples, self.SYN_URI, TG_SYNTHESIS) + + def test_synthesis_derived_from_focus(self): + triples = synthesis_triples(self.SYN_URI, self.FOC_URI) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.SYN_URI) + assert derived is not None + assert derived.o.iri == self.FOC_URI + + def test_synthesis_with_inline_content(self): + triples = synthesis_triples(self.SYN_URI, self.FOC_URI, answer_text="The answer is 42") + content = find_triple(triples, TG_CONTENT, self.SYN_URI) + assert content is not None + assert content.o.value == "The answer is 42" + + def test_synthesis_with_document_reference(self): + triples = synthesis_triples( + self.SYN_URI, self.FOC_URI, + document_id="urn:trustgraph:question:abc/answer", + ) + doc = find_triple(triples, TG_DOCUMENT, self.SYN_URI) + assert doc is not None + assert doc.o.type == IRI + assert doc.o.iri == "urn:trustgraph:question:abc/answer" + + def test_synthesis_document_takes_precedence(self): + """When both document_id and answer_text are provided, document_id wins.""" + triples = synthesis_triples( + self.SYN_URI, self.FOC_URI, + answer_text="inline", + document_id="urn:doc:123", + ) + doc = find_triple(triples, TG_DOCUMENT, self.SYN_URI) + assert doc is not None + content = find_triple(triples, TG_CONTENT, self.SYN_URI) + assert content is None + + def test_synthesis_no_content_or_document(self): + triples = synthesis_triples(self.SYN_URI, self.FOC_URI) + content = find_triple(triples, TG_CONTENT, self.SYN_URI) + doc = find_triple(triples, TG_DOCUMENT, self.SYN_URI) + assert content is None + assert doc is None + + +# --------------------------------------------------------------------------- +# DocumentRAG query-time triples +# --------------------------------------------------------------------------- + +class TestDocRagQuestionTriples: + + Q_URI = "urn:trustgraph:docrag:test-session" + + def test_docrag_question_types(self): + triples = docrag_question_triples(self.Q_URI, "Find info", "2024-01-01T00:00:00Z") + assert has_type(triples, self.Q_URI, PROV_ACTIVITY) + assert has_type(triples, self.Q_URI, TG_QUESTION) + assert has_type(triples, self.Q_URI, TG_DOC_RAG_QUESTION) + + def test_docrag_question_label(self): + triples = docrag_question_triples(self.Q_URI, "Q", "2024-01-01T00:00:00Z") + label = find_triple(triples, RDFS_LABEL, self.Q_URI) + assert label.o.value == "DocumentRAG Question" + + def test_docrag_question_query_text(self): + triples = docrag_question_triples(self.Q_URI, "search query", "2024-01-01T00:00:00Z") + query = find_triple(triples, TG_QUERY, self.Q_URI) + assert query.o.value == "search query" + + +class TestDocRagExplorationTriples: + + EXP_URI = "urn:trustgraph:docrag:test/exploration" + Q_URI = "urn:trustgraph:docrag:test" + + def test_docrag_exploration_types(self): + triples = docrag_exploration_triples(self.EXP_URI, self.Q_URI, 5) + assert has_type(triples, self.EXP_URI, PROV_ENTITY) + assert has_type(triples, self.EXP_URI, TG_EXPLORATION) + + def test_docrag_exploration_generated_by(self): + triples = docrag_exploration_triples(self.EXP_URI, self.Q_URI, 5) + gen = find_triple(triples, PROV_WAS_GENERATED_BY, self.EXP_URI) + assert gen.o.iri == self.Q_URI + + def test_docrag_exploration_chunk_count(self): + triples = docrag_exploration_triples(self.EXP_URI, self.Q_URI, 7) + cc = find_triple(triples, TG_CHUNK_COUNT, self.EXP_URI) + assert cc.o.value == "7" + + def test_docrag_exploration_without_chunk_ids(self): + triples = docrag_exploration_triples(self.EXP_URI, self.Q_URI, 3) + chunks = find_triples(triples, TG_SELECTED_CHUNK) + assert len(chunks) == 0 + + def test_docrag_exploration_with_chunk_ids(self): + chunk_ids = ["urn:chunk:1", "urn:chunk:2", "urn:chunk:3"] + triples = docrag_exploration_triples(self.EXP_URI, self.Q_URI, 3, chunk_ids) + chunks = find_triples(triples, TG_SELECTED_CHUNK, self.EXP_URI) + assert len(chunks) == 3 + chunk_uris = {t.o.iri for t in chunks} + assert chunk_uris == set(chunk_ids) + + +class TestDocRagSynthesisTriples: + + SYN_URI = "urn:trustgraph:docrag:test/synthesis" + EXP_URI = "urn:trustgraph:docrag:test/exploration" + + def test_docrag_synthesis_types(self): + triples = docrag_synthesis_triples(self.SYN_URI, self.EXP_URI) + assert has_type(triples, self.SYN_URI, PROV_ENTITY) + assert has_type(triples, self.SYN_URI, TG_SYNTHESIS) + + def test_docrag_synthesis_derived_from_exploration(self): + """DocRAG skips the focus step — synthesis derives from exploration.""" + triples = docrag_synthesis_triples(self.SYN_URI, self.EXP_URI) + derived = find_triple(triples, PROV_WAS_DERIVED_FROM, self.SYN_URI) + assert derived.o.iri == self.EXP_URI + + def test_docrag_synthesis_with_inline(self): + triples = docrag_synthesis_triples(self.SYN_URI, self.EXP_URI, answer_text="answer") + content = find_triple(triples, TG_CONTENT, self.SYN_URI) + assert content.o.value == "answer" + + def test_docrag_synthesis_with_document(self): + triples = docrag_synthesis_triples( + self.SYN_URI, self.EXP_URI, document_id="urn:doc:ans" + ) + doc = find_triple(triples, TG_DOCUMENT, self.SYN_URI) + assert doc.o.iri == "urn:doc:ans" + content = find_triple(triples, TG_CONTENT, self.SYN_URI) + assert content is None diff --git a/tests/unit/test_provenance/test_uris.py b/tests/unit/test_provenance/test_uris.py new file mode 100644 index 00000000..0e69734c --- /dev/null +++ b/tests/unit/test_provenance/test_uris.py @@ -0,0 +1,292 @@ +""" +Tests for provenance URI generation functions. +""" + +import pytest +from unittest.mock import patch + +from trustgraph.provenance.uris import ( + TRUSTGRAPH_BASE, + _encode_id, + document_uri, + page_uri, + chunk_uri_from_page, + chunk_uri_from_doc, + activity_uri, + subgraph_uri, + agent_uri, + question_uri, + exploration_uri, + focus_uri, + synthesis_uri, + edge_selection_uri, + agent_session_uri, + agent_iteration_uri, + agent_final_uri, + docrag_question_uri, + docrag_exploration_uri, + docrag_synthesis_uri, +) + + +class TestEncodeId: + """Tests for the _encode_id helper.""" + + def test_plain_string(self): + assert _encode_id("abc123") == "abc123" + + def test_string_with_spaces(self): + assert _encode_id("hello world") == "hello%20world" + + def test_string_with_slashes(self): + assert _encode_id("a/b/c") == "a%2Fb%2Fc" + + def test_integer_input(self): + assert _encode_id(42) == "42" + + def test_empty_string(self): + assert _encode_id("") == "" + + def test_special_characters(self): + result = _encode_id("name@domain.com") + assert "@" not in result or result == "name%40domain.com" + + +class TestDocumentUris: + """Tests for document, page, and chunk URI generation.""" + + def test_document_uri_passthrough(self): + iri = "https://example.com/doc/123" + assert document_uri(iri) == iri + + def test_page_uri_format(self): + result = page_uri("https://example.com/doc/123", 5) + assert result == "https://example.com/doc/123/p5" + + def test_page_uri_page_zero(self): + result = page_uri("https://example.com/doc/123", 0) + assert result == "https://example.com/doc/123/p0" + + def test_chunk_uri_from_page_format(self): + result = chunk_uri_from_page("https://example.com/doc/123", 2, 3) + assert result == "https://example.com/doc/123/p2/c3" + + def test_chunk_uri_from_doc_format(self): + result = chunk_uri_from_doc("https://example.com/doc/123", 7) + assert result == "https://example.com/doc/123/c7" + + def test_page_uri_preserves_doc_iri(self): + doc = "urn:isbn:978-3-16-148410-0" + result = page_uri(doc, 1) + assert result.startswith(doc) + + def test_chunk_from_page_hierarchy(self): + """Chunk URI should contain both page and chunk identifiers.""" + result = chunk_uri_from_page("https://example.com/doc", 3, 5) + assert "/p3/" in result + assert result.endswith("/c5") + + +class TestActivityAndSubgraphUris: + """Tests for activity_uri, subgraph_uri, and agent_uri.""" + + def test_activity_uri_with_id(self): + result = activity_uri("my-activity-id") + assert result == f"{TRUSTGRAPH_BASE}/activity/my-activity-id" + + def test_activity_uri_auto_generates_uuid(self): + result = activity_uri() + assert result.startswith(f"{TRUSTGRAPH_BASE}/activity/") + # UUID part should be non-empty + uuid_part = result.split("/activity/")[1] + assert len(uuid_part) > 0 + + def test_activity_uri_unique_uuids(self): + r1 = activity_uri() + r2 = activity_uri() + assert r1 != r2 + + def test_activity_uri_encodes_special_chars(self): + result = activity_uri("id with spaces") + assert "id%20with%20spaces" in result + + def test_subgraph_uri_with_id(self): + result = subgraph_uri("sg-123") + assert result == f"{TRUSTGRAPH_BASE}/subgraph/sg-123" + + def test_subgraph_uri_auto_generates_uuid(self): + result = subgraph_uri() + assert result.startswith(f"{TRUSTGRAPH_BASE}/subgraph/") + uuid_part = result.split("/subgraph/")[1] + assert len(uuid_part) > 0 + + def test_subgraph_uri_unique_uuids(self): + r1 = subgraph_uri() + r2 = subgraph_uri() + assert r1 != r2 + + def test_agent_uri_format(self): + result = agent_uri("pdf-extractor") + assert result == f"{TRUSTGRAPH_BASE}/agent/pdf-extractor" + + def test_agent_uri_encodes_special_chars(self): + result = agent_uri("my component") + assert "my%20component" in result + + +class TestGraphRagQueryUris: + """Tests for GraphRAG query-time provenance URIs.""" + + FIXED_UUID = "550e8400-e29b-41d4-a716-446655440000" + + def test_question_uri_with_session_id(self): + result = question_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:question:{self.FIXED_UUID}" + + def test_question_uri_auto_generates(self): + result = question_uri() + assert result.startswith("urn:trustgraph:question:") + uuid_part = result.split("urn:trustgraph:question:")[1] + assert len(uuid_part) > 0 + + def test_question_uri_unique(self): + r1 = question_uri() + r2 = question_uri() + assert r1 != r2 + + def test_exploration_uri_format(self): + result = exploration_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:prov:exploration:{self.FIXED_UUID}" + + def test_focus_uri_format(self): + result = focus_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:prov:focus:{self.FIXED_UUID}" + + def test_synthesis_uri_format(self): + result = synthesis_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:prov:synthesis:{self.FIXED_UUID}" + + def test_edge_selection_uri_format(self): + result = edge_selection_uri(self.FIXED_UUID, 3) + assert result == f"urn:trustgraph:prov:edge:{self.FIXED_UUID}:3" + + def test_edge_selection_uri_zero_index(self): + result = edge_selection_uri(self.FIXED_UUID, 0) + assert result.endswith(":0") + + def test_session_uris_share_session_id(self): + """All URIs for a session should contain the same session ID.""" + sid = self.FIXED_UUID + q = question_uri(sid) + e = exploration_uri(sid) + f = focus_uri(sid) + s = synthesis_uri(sid) + for uri in [q, e, f, s]: + assert sid in uri + + +class TestAgentProvenanceUris: + """Tests for agent provenance URIs.""" + + FIXED_UUID = "661e8400-e29b-41d4-a716-446655440000" + + def test_agent_session_uri_with_id(self): + result = agent_session_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:agent:{self.FIXED_UUID}" + + def test_agent_session_uri_auto_generates(self): + result = agent_session_uri() + assert result.startswith("urn:trustgraph:agent:") + + def test_agent_session_uri_unique(self): + r1 = agent_session_uri() + r2 = agent_session_uri() + assert r1 != r2 + + def test_agent_iteration_uri_format(self): + result = agent_iteration_uri(self.FIXED_UUID, 1) + assert result == f"urn:trustgraph:agent:{self.FIXED_UUID}/i1" + + def test_agent_iteration_uri_numbering(self): + r1 = agent_iteration_uri(self.FIXED_UUID, 1) + r2 = agent_iteration_uri(self.FIXED_UUID, 2) + assert r1 != r2 + assert r1.endswith("/i1") + assert r2.endswith("/i2") + + def test_agent_final_uri_format(self): + result = agent_final_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:agent:{self.FIXED_UUID}/final" + + def test_agent_uris_share_session_id(self): + sid = self.FIXED_UUID + session = agent_session_uri(sid) + iteration = agent_iteration_uri(sid, 1) + final = agent_final_uri(sid) + for uri in [session, iteration, final]: + assert sid in uri + + +class TestDocRagProvenanceUris: + """Tests for Document RAG provenance URIs.""" + + FIXED_UUID = "772e8400-e29b-41d4-a716-446655440000" + + def test_docrag_question_uri_with_id(self): + result = docrag_question_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:docrag:{self.FIXED_UUID}" + + def test_docrag_question_uri_auto_generates(self): + result = docrag_question_uri() + assert result.startswith("urn:trustgraph:docrag:") + + def test_docrag_question_uri_unique(self): + r1 = docrag_question_uri() + r2 = docrag_question_uri() + assert r1 != r2 + + def test_docrag_exploration_uri_format(self): + result = docrag_exploration_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:docrag:{self.FIXED_UUID}/exploration" + + def test_docrag_synthesis_uri_format(self): + result = docrag_synthesis_uri(self.FIXED_UUID) + assert result == f"urn:trustgraph:docrag:{self.FIXED_UUID}/synthesis" + + def test_docrag_uris_share_session_id(self): + sid = self.FIXED_UUID + q = docrag_question_uri(sid) + e = docrag_exploration_uri(sid) + s = docrag_synthesis_uri(sid) + for uri in [q, e, s]: + assert sid in uri + + +class TestUriNamespaceIsolation: + """Verify that different provenance types use distinct URI namespaces.""" + + FIXED_UUID = "883e8400-e29b-41d4-a716-446655440000" + + def test_graphrag_vs_agent_namespace(self): + graphrag = question_uri(self.FIXED_UUID) + agent = agent_session_uri(self.FIXED_UUID) + assert graphrag != agent + assert "question" in graphrag + assert "agent" in agent + + def test_graphrag_vs_docrag_namespace(self): + graphrag = question_uri(self.FIXED_UUID) + docrag = docrag_question_uri(self.FIXED_UUID) + assert graphrag != docrag + + def test_agent_vs_docrag_namespace(self): + agent = agent_session_uri(self.FIXED_UUID) + docrag = docrag_question_uri(self.FIXED_UUID) + assert agent != docrag + + def test_extraction_vs_query_namespace(self): + """Extraction URIs use https://, query URIs use urn:.""" + ext = activity_uri(self.FIXED_UUID) + query = question_uri(self.FIXED_UUID) + assert ext.startswith("https://") + assert query.startswith("urn:") diff --git a/tests/unit/test_provenance/test_vocabulary.py b/tests/unit/test_provenance/test_vocabulary.py new file mode 100644 index 00000000..a3c644e8 --- /dev/null +++ b/tests/unit/test_provenance/test_vocabulary.py @@ -0,0 +1,124 @@ +""" +Tests for provenance vocabulary bootstrap. +""" + +import pytest + +from trustgraph.schema import Triple, Term, IRI, LITERAL + +from trustgraph.provenance.vocabulary import ( + get_vocabulary_triples, + PROV_CLASS_LABELS, + PROV_PREDICATE_LABELS, + DC_PREDICATE_LABELS, + SCHEMA_LABELS, + SKOS_LABELS, + TG_CLASS_LABELS, + TG_PREDICATE_LABELS, +) + +from trustgraph.provenance.namespaces import ( + RDFS_LABEL, + PROV_ENTITY, PROV_ACTIVITY, PROV_AGENT, + PROV_WAS_DERIVED_FROM, PROV_WAS_GENERATED_BY, + PROV_USED, PROV_WAS_ASSOCIATED_WITH, PROV_STARTED_AT_TIME, + DC_TITLE, DC_SOURCE, DC_DATE, DC_CREATOR, + TG_DOCUMENT_TYPE, TG_PAGE_TYPE, TG_CHUNK_TYPE, TG_SUBGRAPH_TYPE, +) + + +class TestVocabularyTriples: + """Tests for the vocabulary bootstrap function.""" + + def test_returns_list_of_triples(self): + result = get_vocabulary_triples() + assert isinstance(result, list) + assert len(result) > 0 + for t in result: + assert isinstance(t, Triple) + + def test_all_triples_are_label_triples(self): + """Every vocabulary triple should use rdfs:label as predicate.""" + for t in get_vocabulary_triples(): + assert t.p.type == IRI + assert t.p.iri == RDFS_LABEL + + def test_all_subjects_are_iris(self): + for t in get_vocabulary_triples(): + assert t.s.type == IRI + assert len(t.s.iri) > 0 + + def test_all_objects_are_literals(self): + for t in get_vocabulary_triples(): + assert t.o.type == LITERAL + assert len(t.o.value) > 0 + + def test_no_duplicate_subjects(self): + subjects = [t.s.iri for t in get_vocabulary_triples()] + assert len(subjects) == len(set(subjects)) + + def test_includes_prov_classes(self): + subjects = {t.s.iri for t in get_vocabulary_triples()} + assert PROV_ENTITY in subjects + assert PROV_ACTIVITY in subjects + assert PROV_AGENT in subjects + + def test_includes_prov_predicates(self): + subjects = {t.s.iri for t in get_vocabulary_triples()} + assert PROV_WAS_DERIVED_FROM in subjects + assert PROV_WAS_GENERATED_BY in subjects + assert PROV_USED in subjects + assert PROV_WAS_ASSOCIATED_WITH in subjects + assert PROV_STARTED_AT_TIME in subjects + + def test_includes_dc_predicates(self): + subjects = {t.s.iri for t in get_vocabulary_triples()} + assert DC_TITLE in subjects + assert DC_SOURCE in subjects + assert DC_DATE in subjects + assert DC_CREATOR in subjects + + def test_includes_tg_classes(self): + subjects = {t.s.iri for t in get_vocabulary_triples()} + assert TG_DOCUMENT_TYPE in subjects + assert TG_PAGE_TYPE in subjects + assert TG_CHUNK_TYPE in subjects + assert TG_SUBGRAPH_TYPE in subjects + + def test_component_lists_sum_to_total(self): + total = get_vocabulary_triples() + components = ( + PROV_CLASS_LABELS + + PROV_PREDICATE_LABELS + + DC_PREDICATE_LABELS + + SCHEMA_LABELS + + SKOS_LABELS + + TG_CLASS_LABELS + + TG_PREDICATE_LABELS + ) + assert len(total) == len(components) + + def test_idempotent(self): + """Calling twice should return equivalent triples.""" + r1 = get_vocabulary_triples() + r2 = get_vocabulary_triples() + assert len(r1) == len(r2) + for t1, t2 in zip(r1, r2): + assert t1.s.iri == t2.s.iri + assert t1.o.value == t2.o.value + + +class TestNamespaceConstants: + """Verify namespace constants are well-formed IRIs.""" + + def test_prov_namespace_prefix(self): + assert PROV_ENTITY.startswith("http://www.w3.org/ns/prov#") + + def test_dc_namespace_prefix(self): + assert DC_TITLE.startswith("http://purl.org/dc/elements/1.1/") + + def test_tg_namespace_prefix(self): + assert TG_DOCUMENT_TYPE.startswith("https://trustgraph.ai/ns/") + + def test_rdfs_label_iri(self): + assert RDFS_LABEL == "http://www.w3.org/2000/01/rdf-schema#label"