apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-23 20:21:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Tests

Browse source
This commit is contained in:
Cyber MacGeddon 2026-01-27 09:20:01 +00:00
parent 3af813bf13
commit 325deb2694
2 changed files with 147 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/unit/test_knowledge_graph/test_graph_validation.py
View file

@ -7,7 +7,7 @@ processing graph structures, and performing graph operations.
import pytest import pytest
from unittest.mock import Mock from unittest.mock import Mock
from .conftest import Triple, Value, Metadata from .conftest import Triple, Metadata
from collections import defaultdict, deque from collections import defaultdict, deque

274
tests/unit/test_knowledge_graph/test_triple_construction.py
View file

@ -2,13 +2,13 @@
Unit tests for triple construction logic Unit tests for triple construction logic
Tests the core business logic for constructing RDF triples from extracted Tests the core business logic for constructing RDF triples from extracted
entities and relationships, including URI generation, Value object creation, entities and relationships, including URI generation, Term object creation,
and triple validation. and triple validation.
""" """
import pytest import pytest
from unittest.mock import Mock from unittest.mock import Mock
from .conftest import Triple, Triples, Value, Metadata from .conftest import Triple, Triples, Term, Metadata, IRI, LITERAL
import re import re
import hashlib import hashlib
@ -48,80 +48,82 @@ class TestTripleConstructionLogic:
generated_uri = generate_uri(text, entity_type) generated_uri = generate_uri(text, entity_type)
assert generated_uri == expected_uri, f"URI generation failed for '{text}'" assert generated_uri == expected_uri, f"URI generation failed for '{text}'"
def test_value_object_creation(self): def test_term_object_creation(self):
"""Test creation of Value objects for subjects, predicates, and objects""" """Test creation of Term objects for subjects, predicates, and objects"""
# Arrange # Arrange
def create_value_object(text, is_uri, value_type=""): def create_term_object(text, is_uri, datatype=""):
return Value( if is_uri:
value=text, return Term(type=IRI, iri=text)
is_uri=is_uri, else:
type=value_type return Term(type=LITERAL, value=text, datatype=datatype if datatype else None)
)
test_cases = [ test_cases = [
("http://trustgraph.ai/kg/person/john-smith", True, ""), ("http://trustgraph.ai/kg/person/john-smith", True, ""),
("John Smith", False, "string"), ("John Smith", False, "string"),
("42", False, "integer"), ("42", False, "integer"),
("http://schema.org/worksFor", True, "") ("http://schema.org/worksFor", True, "")
] ]
# Act & Assert # Act & Assert
for value_text, is_uri, value_type in test_cases: for value_text, is_uri, datatype in test_cases:
value_obj = create_value_object(value_text, is_uri, value_type) term_obj = create_term_object(value_text, is_uri, datatype)
assert isinstance(value_obj, Value) assert isinstance(term_obj, Term)
assert value_obj.value == value_text if is_uri:
assert value_obj.is_uri == is_uri assert term_obj.type == IRI
assert value_obj.type == value_type assert term_obj.iri == value_text
else:
assert term_obj.type == LITERAL
assert term_obj.value == value_text
def test_triple_construction_from_relationship(self): def test_triple_construction_from_relationship(self):
"""Test constructing Triple objects from relationships""" """Test constructing Triple objects from relationships"""
# Arrange # Arrange
relationship = { relationship = {
"subject": "John Smith", "subject": "John Smith",
"predicate": "works_for", "predicate": "works_for",
"object": "OpenAI", "object": "OpenAI",
"subject_type": "PERSON", "subject_type": "PERSON",
"object_type": "ORG" "object_type": "ORG"
} }
def construct_triple(relationship, uri_base="http://trustgraph.ai/kg"): def construct_triple(relationship, uri_base="http://trustgraph.ai/kg"):
# Generate URIs # Generate URIs
subject_uri = f"{uri_base}/person/{relationship['subject'].lower().replace(' ', '-')}" subject_uri = f"{uri_base}/person/{relationship['subject'].lower().replace(' ', '-')}"
object_uri = f"{uri_base}/org/{relationship['object'].lower().replace(' ', '-')}" object_uri = f"{uri_base}/org/{relationship['object'].lower().replace(' ', '-')}"
# Map predicate to schema.org URI # Map predicate to schema.org URI
predicate_mappings = { predicate_mappings = {
"works_for": "http://schema.org/worksFor", "works_for": "http://schema.org/worksFor",
"located_in": "http://schema.org/location", "located_in": "http://schema.org/location",
"developed": "http://schema.org/creator" "developed": "http://schema.org/creator"
} }
predicate_uri = predicate_mappings.get(relationship["predicate"], predicate_uri = predicate_mappings.get(relationship["predicate"],
f"{uri_base}/predicate/{relationship['predicate']}") f"{uri_base}/predicate/{relationship['predicate']}")
# Create Value objects # Create Term objects
subject_value = Value(value=subject_uri, is_uri=True, type="") subject_term = Term(type=IRI, iri=subject_uri)
predicate_value = Value(value=predicate_uri, is_uri=True, type="") predicate_term = Term(type=IRI, iri=predicate_uri)
object_value = Value(value=object_uri, is_uri=True, type="") object_term = Term(type=IRI, iri=object_uri)
# Create Triple # Create Triple
return Triple( return Triple(
s=subject_value, s=subject_term,
p=predicate_value, p=predicate_term,
o=object_value o=object_term
) )
# Act # Act
triple = construct_triple(relationship) triple = construct_triple(relationship)
# Assert # Assert
assert isinstance(triple, Triple) assert isinstance(triple, Triple)
assert triple.s.value == "http://trustgraph.ai/kg/person/john-smith" assert triple.s.iri == "http://trustgraph.ai/kg/person/john-smith"
assert triple.s.is_uri is True assert triple.s.type == IRI
assert triple.p.value == "http://schema.org/worksFor" assert triple.p.iri == "http://schema.org/worksFor"
assert triple.p.is_uri is True assert triple.p.type == IRI
assert triple.o.value == "http://trustgraph.ai/kg/org/openai" assert triple.o.iri == "http://trustgraph.ai/kg/org/openai"
assert triple.o.is_uri is True assert triple.o.type == IRI
def test_literal_value_handling(self): def test_literal_value_handling(self):
"""Test handling of literal values vs URI values""" """Test handling of literal values vs URI values"""
@ -132,10 +134,10 @@ class TestTripleConstructionLogic:
("John Smith", "email", "john@example.com", False), # Literal email ("John Smith", "email", "john@example.com", False), # Literal email
("John Smith", "worksFor", "http://trustgraph.ai/kg/org/openai", True) # URI reference ("John Smith", "worksFor", "http://trustgraph.ai/kg/org/openai", True) # URI reference
] ]
def create_triple_with_literal(subject_uri, predicate, object_value, object_is_uri): def create_triple_with_literal(subject_uri, predicate, object_value, object_is_uri):
subject_val = Value(value=subject_uri, is_uri=True, type="") subject_term = Term(type=IRI, iri=subject_uri)
# Determine predicate URI # Determine predicate URI
predicate_mappings = { predicate_mappings = {
"name": "http://schema.org/name", "name": "http://schema.org/name",
@ -144,32 +146,37 @@ class TestTripleConstructionLogic:
"worksFor": "http://schema.org/worksFor" "worksFor": "http://schema.org/worksFor"
} }
predicate_uri = predicate_mappings.get(predicate, f"http://trustgraph.ai/kg/predicate/{predicate}") predicate_uri = predicate_mappings.get(predicate, f"http://trustgraph.ai/kg/predicate/{predicate}")
predicate_val = Value(value=predicate_uri, is_uri=True, type="") predicate_term = Term(type=IRI, iri=predicate_uri)
# Create object value with appropriate type # Create object term with appropriate type
object_type = "" if object_is_uri:
if not object_is_uri: object_term = Term(type=IRI, iri=object_value)
else:
datatype = None
if predicate == "age": if predicate == "age":
object_type = "integer" datatype = "integer"
elif predicate in ["name", "email"]: elif predicate in ["name", "email"]:
object_type = "string" datatype = "string"
object_term = Term(type=LITERAL, value=object_value, datatype=datatype)
object_val = Value(value=object_value, is_uri=object_is_uri, type=object_type)
return Triple(s=subject_term, p=predicate_term, o=object_term)
return Triple(s=subject_val, p=predicate_val, o=object_val)
# Act & Assert # Act & Assert
for subject_uri, predicate, object_value, object_is_uri in test_data: for subject_uri, predicate, object_value, object_is_uri in test_data:
subject_full_uri = "http://trustgraph.ai/kg/person/john-smith" subject_full_uri = "http://trustgraph.ai/kg/person/john-smith"
triple = create_triple_with_literal(subject_full_uri, predicate, object_value, object_is_uri) triple = create_triple_with_literal(subject_full_uri, predicate, object_value, object_is_uri)
assert triple.o.is_uri == object_is_uri if object_is_uri:
assert triple.o.value == object_value assert triple.o.type == IRI
assert triple.o.iri == object_value
else:
assert triple.o.type == LITERAL
assert triple.o.value == object_value
if predicate == "age": if predicate == "age":
assert triple.o.type == "integer" assert triple.o.datatype == "integer"
elif predicate in ["name", "email"]: elif predicate in ["name", "email"]:
assert triple.o.type == "string" assert triple.o.datatype == "string"
def test_namespace_management(self): def test_namespace_management(self):
"""Test namespace prefix management and expansion""" """Test namespace prefix management and expansion"""
@ -216,63 +223,74 @@ class TestTripleConstructionLogic:
def test_triple_validation(self): def test_triple_validation(self):
"""Test triple validation rules""" """Test triple validation rules"""
# Arrange # Arrange
def get_term_value(term):
"""Extract value from a Term"""
if term.type == IRI:
return term.iri
else:
return term.value
def validate_triple(triple): def validate_triple(triple):
errors = [] errors = []
# Check required components # Check required components
if not triple.s or not triple.s.value: s_val = get_term_value(triple.s) if triple.s else None
p_val = get_term_value(triple.p) if triple.p else None
o_val = get_term_value(triple.o) if triple.o else None
if not triple.s or not s_val:
errors.append("Missing or empty subject") errors.append("Missing or empty subject")
if not triple.p or not triple.p.value: if not triple.p or not p_val:
errors.append("Missing or empty predicate") errors.append("Missing or empty predicate")
if not triple.o or not triple.o.value: if not triple.o or not o_val:
errors.append("Missing or empty object") errors.append("Missing or empty object")
# Check URI validity for URI values # Check URI validity for URI values
uri_pattern = r'^https?://[^\s/$.?#].[^\s]*$' uri_pattern = r'^https?://[^\s/$.?#].[^\s]*$'
if triple.s.is_uri and not re.match(uri_pattern, triple.s.value): if triple.s.type == IRI and not re.match(uri_pattern, triple.s.iri or ""):
errors.append("Invalid subject URI format") errors.append("Invalid subject URI format")
if triple.p.is_uri and not re.match(uri_pattern, triple.p.value): if triple.p.type == IRI and not re.match(uri_pattern, triple.p.iri or ""):
errors.append("Invalid predicate URI format") errors.append("Invalid predicate URI format")
if triple.o.is_uri and not re.match(uri_pattern, triple.o.value): if triple.o.type == IRI and not re.match(uri_pattern, triple.o.iri or ""):
errors.append("Invalid object URI format") errors.append("Invalid object URI format")
# Predicates should typically be URIs # Predicates should typically be URIs
if not triple.p.is_uri: if triple.p.type != IRI:
errors.append("Predicate should be a URI") errors.append("Predicate should be a URI")
return len(errors) == 0, errors return len(errors) == 0, errors
# Test valid triple # Test valid triple
valid_triple = Triple( valid_triple = Triple(
s=Value(value="http://trustgraph.ai/kg/person/john", is_uri=True, type=""), s=Term(type=IRI, iri="http://trustgraph.ai/kg/person/john"),
p=Value(value="http://schema.org/name", is_uri=True, type=""), p=Term(type=IRI, iri="http://schema.org/name"),
o=Value(value="John Smith", is_uri=False, type="string") o=Term(type=LITERAL, value="John Smith", datatype="string")
) )
# Test invalid triples # Test invalid triples
invalid_triples = [ invalid_triples = [
Triple(s=Value(value="", is_uri=True, type=""), Triple(s=Term(type=IRI, iri=""),
p=Value(value="http://schema.org/name", is_uri=True, type=""), p=Term(type=IRI, iri="http://schema.org/name"),
o=Value(value="John", is_uri=False, type="")), # Empty subject o=Term(type=LITERAL, value="John")), # Empty subject
Triple(s=Value(value="http://trustgraph.ai/kg/person/john", is_uri=True, type=""), Triple(s=Term(type=IRI, iri="http://trustgraph.ai/kg/person/john"),
p=Value(value="name", is_uri=False, type=""), # Non-URI predicate p=Term(type=LITERAL, value="name"), # Non-URI predicate
o=Value(value="John", is_uri=False, type="")), o=Term(type=LITERAL, value="John")),
Triple(s=Value(value="invalid-uri", is_uri=True, type=""), Triple(s=Term(type=IRI, iri="invalid-uri"),
p=Value(value="http://schema.org/name", is_uri=True, type=""), p=Term(type=IRI, iri="http://schema.org/name"),
o=Value(value="John", is_uri=False, type="")) # Invalid URI format o=Term(type=LITERAL, value="John")) # Invalid URI format
] ]
# Act & Assert # Act & Assert
is_valid, errors = validate_triple(valid_triple) is_valid, errors = validate_triple(valid_triple)
assert is_valid, f"Valid triple failed validation: {errors}" assert is_valid, f"Valid triple failed validation: {errors}"
for invalid_triple in invalid_triples: for invalid_triple in invalid_triples:
is_valid, errors = validate_triple(invalid_triple) is_valid, errors = validate_triple(invalid_triple)
assert not is_valid, f"Invalid triple passed validation: {invalid_triple}" assert not is_valid, f"Invalid triple passed validation: {invalid_triple}"
@ -286,97 +304,97 @@ class TestTripleConstructionLogic:
{"text": "OpenAI", "type": "ORG"}, {"text": "OpenAI", "type": "ORG"},
{"text": "San Francisco", "type": "PLACE"} {"text": "San Francisco", "type": "PLACE"}
] ]
relationships = [ relationships = [
{"subject": "John Smith", "predicate": "works_for", "object": "OpenAI"}, {"subject": "John Smith", "predicate": "works_for", "object": "OpenAI"},
{"subject": "OpenAI", "predicate": "located_in", "object": "San Francisco"} {"subject": "OpenAI", "predicate": "located_in", "object": "San Francisco"}
] ]
def construct_triple_batch(entities, relationships, document_id="doc-1"): def construct_triple_batch(entities, relationships, document_id="doc-1"):
triples = [] triples = []
# Create type triples for entities # Create type triples for entities
for entity in entities: for entity in entities:
entity_uri = f"http://trustgraph.ai/kg/{entity['type'].lower()}/{entity['text'].lower().replace(' ', '-')}" entity_uri = f"http://trustgraph.ai/kg/{entity['type'].lower()}/{entity['text'].lower().replace(' ', '-')}"
type_uri = f"http://trustgraph.ai/kg/type/{entity['type']}" type_uri = f"http://trustgraph.ai/kg/type/{entity['type']}"
type_triple = Triple( type_triple = Triple(
s=Value(value=entity_uri, is_uri=True, type=""), s=Term(type=IRI, iri=entity_uri),
p=Value(value="http://www.w3.org/1999/02/22-rdf-syntax-ns#type", is_uri=True, type=""), p=Term(type=IRI, iri="http://www.w3.org/1999/02/22-rdf-syntax-ns#type"),
o=Value(value=type_uri, is_uri=True, type="") o=Term(type=IRI, iri=type_uri)
) )
triples.append(type_triple) triples.append(type_triple)
# Create relationship triples # Create relationship triples
for rel in relationships: for rel in relationships:
subject_uri = f"http://trustgraph.ai/kg/entity/{rel['subject'].lower().replace(' ', '-')}" subject_uri = f"http://trustgraph.ai/kg/entity/{rel['subject'].lower().replace(' ', '-')}"
object_uri = f"http://trustgraph.ai/kg/entity/{rel['object'].lower().replace(' ', '-')}" object_uri = f"http://trustgraph.ai/kg/entity/{rel['object'].lower().replace(' ', '-')}"
predicate_uri = f"http://schema.org/{rel['predicate'].replace('_', '')}" predicate_uri = f"http://schema.org/{rel['predicate'].replace('_', '')}"
rel_triple = Triple( rel_triple = Triple(
s=Value(value=subject_uri, is_uri=True, type=""), s=Term(type=IRI, iri=subject_uri),
p=Value(value=predicate_uri, is_uri=True, type=""), p=Term(type=IRI, iri=predicate_uri),
o=Value(value=object_uri, is_uri=True, type="") o=Term(type=IRI, iri=object_uri)
) )
triples.append(rel_triple) triples.append(rel_triple)
return triples return triples
# Act # Act
triples = construct_triple_batch(entities, relationships) triples = construct_triple_batch(entities, relationships)
# Assert # Assert
assert len(triples) == len(entities) + len(relationships) # Type triples + relationship triples assert len(triples) == len(entities) + len(relationships) # Type triples + relationship triples
# Check that all triples are valid Triple objects # Check that all triples are valid Triple objects
for triple in triples: for triple in triples:
assert isinstance(triple, Triple) assert isinstance(triple, Triple)
assert triple.s.value != "" assert triple.s.iri != ""
assert triple.p.value != "" assert triple.p.iri != ""
assert triple.o.value != "" assert triple.o.iri != ""
def test_triples_batch_object_creation(self): def test_triples_batch_object_creation(self):
"""Test creating Triples batch objects with metadata""" """Test creating Triples batch objects with metadata"""
# Arrange # Arrange
sample_triples = [ sample_triples = [
Triple( Triple(
s=Value(value="http://trustgraph.ai/kg/person/john", is_uri=True, type=""), s=Term(type=IRI, iri="http://trustgraph.ai/kg/person/john"),
p=Value(value="http://schema.org/name", is_uri=True, type=""), p=Term(type=IRI, iri="http://schema.org/name"),
o=Value(value="John Smith", is_uri=False, type="string") o=Term(type=LITERAL, value="John Smith", datatype="string")
), ),
Triple( Triple(
s=Value(value="http://trustgraph.ai/kg/person/john", is_uri=True, type=""), s=Term(type=IRI, iri="http://trustgraph.ai/kg/person/john"),
p=Value(value="http://schema.org/worksFor", is_uri=True, type=""), p=Term(type=IRI, iri="http://schema.org/worksFor"),
o=Value(value="http://trustgraph.ai/kg/org/openai", is_uri=True, type="") o=Term(type=IRI, iri="http://trustgraph.ai/kg/org/openai")
) )
] ]
metadata = Metadata( metadata = Metadata(
id="test-doc-123", id="test-doc-123",
user="test_user", user="test_user",
collection="test_collection", collection="test_collection",
metadata=[] metadata=[]
) )
# Act # Act
triples_batch = Triples( triples_batch = Triples(
metadata=metadata, metadata=metadata,
triples=sample_triples triples=sample_triples
) )
# Assert # Assert
assert isinstance(triples_batch, Triples) assert isinstance(triples_batch, Triples)
assert triples_batch.metadata.id == "test-doc-123" assert triples_batch.metadata.id == "test-doc-123"
assert triples_batch.metadata.user == "test_user" assert triples_batch.metadata.user == "test_user"
assert triples_batch.metadata.collection == "test_collection" assert triples_batch.metadata.collection == "test_collection"
assert len(triples_batch.triples) == 2 assert len(triples_batch.triples) == 2
# Check that triples are properly embedded # Check that triples are properly embedded
for triple in triples_batch.triples: for triple in triples_batch.triples:
assert isinstance(triple, Triple) assert isinstance(triple, Triple)
assert isinstance(triple.s, Value) assert isinstance(triple.s, Term)
assert isinstance(triple.p, Value) assert isinstance(triple.p, Term)
assert isinstance(triple.o, Value) assert isinstance(triple.o, Term)
def test_uri_collision_handling(self): def test_uri_collision_handling(self):
"""Test handling of URI collisions and duplicate detection""" """Test handling of URI collisions and duplicate detection"""