import json
import pytest
from silk import GraphStore
ONTOLOGY = json.dumps({
"node_types": {
"person": {
"properties": {
"name": {"value_type": "string", "required": True},
"age": {"value_type": "int", "required": False},
},
"subtypes": {
"employee": {
"properties": {
"department": {"value_type": "string", "required": True},
}
}
}
},
"document": {
"properties": {
"title": {"value_type": "string", "required": True},
}
}
},
"edge_types": {
"AUTHORED": {
"source_types": ["person"],
"target_types": ["document"],
"properties": {}
}
}
})
def _store():
return GraphStore("test-open", ONTOLOGY)
def test_unknown_property_on_node():
store = _store()
store.add_node("alice", "person", "Alice", {
"name": "Alice",
"email": "alice@example.com", "verified": True, }, subtype="user") node = store.get_node("alice")
assert node["properties"]["name"] == "Alice"
assert node["properties"]["email"] == "alice@example.com"
assert node["properties"]["verified"] is True
def test_unknown_property_on_edge():
store = _store()
store.add_node("alice", "person", "Alice", {"name": "Alice"}, subtype="user")
store.add_node("doc1", "document", "Doc", {"title": "Paper"})
store.add_edge("e1", "AUTHORED", "alice", "doc1", {
"year": 2026, "role": "primary", })
edge = store.get_edge("e1")
assert edge["properties"]["year"] == 2026
assert edge["properties"]["role"] == "primary"
def test_unknown_property_with_known_subtype():
store = _store()
store.add_node("bob", "person", "Bob", {
"name": "Bob",
"department": "eng", "slack_handle": "@bob", }, subtype="employee")
node = store.get_node("bob")
assert node["properties"]["department"] == "eng"
assert node["properties"]["slack_handle"] == "@bob"
def test_unknown_property_survives_sync():
store_a = GraphStore("a", ONTOLOGY)
store_b = GraphStore("b", ONTOLOGY)
store_a.add_node("alice", "person", "Alice", {
"name": "Alice",
"custom_field": "custom_value",
}, subtype="user")
offer = store_a.generate_sync_offer()
payload = store_b.receive_sync_offer(offer)
store_a.merge_sync_payload(payload)
offer = store_b.generate_sync_offer()
payload = store_a.receive_sync_offer(offer)
store_b.merge_sync_payload(payload)
node_b = store_b.get_node("alice")
assert node_b is not None
assert node_b["properties"]["custom_field"] == "custom_value"
def test_unknown_property_survives_persistence(tmp_path):
db_path = str(tmp_path / "test.redb")
store = GraphStore("n1", ONTOLOGY, path=db_path)
store.add_node("alice", "person", "Alice", {
"name": "Alice",
"custom": "persisted",
}, subtype="user")
del store
store2 = GraphStore.open(db_path)
node = store2.get_node("alice")
assert node["properties"]["custom"] == "persisted"
def test_unknown_subtype_accepted():
store = _store()
store.add_node("carol", "person", "Carol", {
"name": "Carol", }, subtype="contractor")
node = store.get_node("carol")
assert node["subtype"] == "contractor"
assert node["properties"]["name"] == "Carol"
def test_unknown_subtype_still_enforces_required():
store = _store()
with pytest.raises(ValueError, match="name"):
store.add_node("x", "person", "X", {}, subtype="contractor")
def test_subtype_on_type_without_subtypes():
store = _store()
store.add_node("doc1", "document", "Doc", {
"title": "Paper",
}, subtype="report")
node = store.get_node("doc1")
assert node["subtype"] == "report"
def test_unknown_subtype_with_extra_properties():
store = _store()
store.add_node("dave", "person", "Dave", {
"name": "Dave",
"contract_end": "2026-12-31", "rate": 150, }, subtype="freelancer") node = store.get_node("dave")
assert node["subtype"] == "freelancer"
assert node["properties"]["contract_end"] == "2026-12-31"
assert node["properties"]["rate"] == 150
def test_required_property_still_enforced():
store = _store()
with pytest.raises(ValueError, match="title"):
store.add_node("x", "document", "X", {})
def test_known_property_type_still_enforced():
store = _store()
with pytest.raises(ValueError, match="age"):
store.add_node("x", "person", "X", {"name": "X", "age": "thirty"}, subtype="user")
def test_known_subtype_required_property_still_enforced():
store = _store()
with pytest.raises(ValueError, match="department"):
store.add_node("x", "person", "X", {"name": "X"}, subtype="employee")
def test_edge_type_constraints_still_enforced():
store = _store()
store.add_node("alice", "person", "Alice", {"name": "Alice"}, subtype="user")
store.add_node("bob", "person", "Bob", {"name": "Bob"}, subtype="user")
with pytest.raises(ValueError):
store.add_edge("e1", "AUTHORED", "alice", "bob")
def test_unknown_node_type_still_rejected():
store = _store()
with pytest.raises(ValueError, match="node type"):
store.add_node("x", "spaceship", "X", {})