generated from coulomb/repo-seed
62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
from kontextual_engine import (
|
|
Artifact,
|
|
ArtifactMetadata,
|
|
Collection,
|
|
DuplicateResourceError,
|
|
InMemoryKnowledgeRepository,
|
|
QueryEngine,
|
|
Relationship,
|
|
)
|
|
|
|
|
|
def test_in_memory_repository_stores_collections_artifacts_and_relationships() -> None:
|
|
repo = InMemoryKnowledgeRepository()
|
|
collection = repo.save_collection(Collection.create("runtime"))
|
|
source = repo.save_artifact(Artifact.create(collection.id, "source", "source text"))
|
|
derived = repo.save_artifact(Artifact.create(collection.id, "derived", "derived text"))
|
|
relationship = repo.save_relationship(Relationship.create(derived.id, source.id, "derived from"))
|
|
|
|
assert repo.get_collection(collection.id).name == "runtime"
|
|
assert repo.get_artifact_by_name(collection.id, "source").id == source.id
|
|
assert repo.list_relationships(artifact_id=source.id) == [relationship]
|
|
|
|
|
|
def test_repository_rejects_duplicate_artifact_names_in_collection() -> None:
|
|
repo = InMemoryKnowledgeRepository()
|
|
collection = repo.save_collection(Collection.create("runtime"))
|
|
repo.save_artifact(Artifact.create(collection.id, "same", "one"))
|
|
|
|
try:
|
|
repo.save_artifact(Artifact.create(collection.id, "same", "two"))
|
|
except DuplicateResourceError as exc:
|
|
assert exc.details["name"] == "same"
|
|
else:
|
|
raise AssertionError("Expected duplicate artifact name to fail")
|
|
|
|
|
|
def test_query_engine_filters_artifacts_and_related_artifacts() -> None:
|
|
repo = InMemoryKnowledgeRepository()
|
|
collection = repo.save_collection(Collection.create("runtime"))
|
|
source = repo.save_artifact(
|
|
Artifact.create(
|
|
collection.id,
|
|
"source",
|
|
"alpha content",
|
|
metadata=ArtifactMetadata(tags=["seed"], custom={"kind": "source"}),
|
|
)
|
|
)
|
|
derived = repo.save_artifact(Artifact.create(collection.id, "derived", "beta content"))
|
|
repo.save_relationship(Relationship.create(derived.id, source.id, "derived from"))
|
|
|
|
query = QueryEngine(repo)
|
|
by_text = query.artifacts(text_contains="alpha")
|
|
by_metadata = query.artifacts(metadata={"custom.kind": "source"})
|
|
related = query.related_artifacts(source.id)
|
|
|
|
assert by_text.result_count == 1
|
|
assert by_text.results[0]["id"] == source.id
|
|
assert by_metadata.result_count == 1
|
|
assert related.result_count == 1
|
|
assert related.results[0]["id"] == derived.id
|
|
|