Files
ThothII/harness/tests/test_qdrant_vector_store.py
T

701 lines
26 KiB
Python

from uuid import NAMESPACE_URL, uuid5
import pytest
import requests
from qdrant_test_helpers import FakeQdrantHttp, FakeResponse, _write_record
from tht.adapters.vector.qdrant import QdrantVectorStore, point_id
from tht.ports.vector import (
SemanticIndexIncompatibleError,
VectorResponseError,
VectorStoreError,
VectorTransportError,
)
_REQUIRED_INDEXES = {
"content_hash", "document_id", "kind", "record_key", "record_kind",
"vector_generation", "workspace_id", "workspace_revision",
}
def _store(fake: FakeQdrantHttp, *, collection_lifecycle="create_if_missing") -> QdrantVectorStore:
return QdrantVectorStore(
base_url="http://qdrant:6333",
collection="workspace-semantic",
workspace_id="demo",
workspace_revision="a" * 40,
expected_dimension=1024,
collection_lifecycle=collection_lifecycle,
request=fake.request,
)
def test_point_id_is_deterministic_uuidv5():
assert point_id("demo", "memory", "memory:1") == str(
uuid5(NAMESPACE_URL, "thothii:demo:memory:memory:1")
)
def test_require_existing_requires_an_explicit_embedding_dimension():
fake = FakeQdrantHttp()
with pytest.raises(ValueError, match="expected dimension"):
QdrantVectorStore(
base_url="http://qdrant:6333",
collection="workspace-semantic",
workspace_id="demo",
expected_dimension=None,
collection_lifecycle="require_existing",
request=fake.request,
)
assert fake.calls == []
def test_require_existing_refuses_missing_collection_without_mutations():
fake = FakeQdrantHttp()
store = _store(fake, collection_lifecycle="require_existing")
with pytest.raises(VectorStoreError, match="semantic_index_incompatible"):
store.upsert("memory", [_write_record("memory:1", "memory")])
assert [call for call in fake.calls if call[0] == "PUT"] == []
def test_require_existing_maps_scroll_404_after_compatible_preflight():
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
fake.payload_indexes = {
"content_hash", "document_id", "kind", "record_key", "record_kind",
"vector_generation", "workspace_id", "workspace_revision",
}
original_request = fake.request
def request(method, url, **kwargs):
if method == "POST" and url.endswith("/points/scroll"):
original_request(method, url, **kwargs)
return FakeResponse(404, {"status": "error"})
return original_request(method, url, **kwargs)
store = QdrantVectorStore(
base_url="http://qdrant:6333", collection="workspace-semantic", workspace_id="demo",
workspace_revision="a" * 40, expected_dimension=1024,
collection_lifecycle="require_existing", request=request,
)
with pytest.raises(SemanticIndexIncompatibleError):
store.existing_hashes("memory", ["memory"])
assert [call for call in fake.calls if call[1].endswith("/points/scroll")]
assert [call for call in fake.calls if call[0] == "PUT"] == []
@pytest.mark.parametrize(
("operation", "generation"),
[("delete_kinds", None), ("delete_generation", "gen:" + "a" * 32)],
)
def test_require_existing_maps_delete_scroll_404_after_compatible_preflight(operation, generation):
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
fake.payload_indexes = {
"content_hash", "document_id", "kind", "record_key", "record_kind",
"vector_generation", "workspace_id", "workspace_revision",
}
original_request = fake.request
def request(method, url, **kwargs):
if method == "POST" and url.endswith("/points/scroll"):
original_request(method, url, **kwargs)
return FakeResponse(404, {"status": "error"})
return original_request(method, url, **kwargs)
store = QdrantVectorStore(
base_url="http://qdrant:6333", collection="workspace-semantic", workspace_id="demo",
workspace_revision="a" * 40, expected_dimension=1024,
collection_lifecycle="require_existing", request=request,
)
with pytest.raises(SemanticIndexIncompatibleError):
if operation == "delete_kinds":
store.delete_kinds("memory", ["memory"])
else:
store.delete_generation("evidence", generation, "demo")
assert [call for call in fake.calls if call[1].endswith("/points/scroll")]
assert [call for call in fake.calls if call[0] == "POST" and "delete" in call[1]] == []
def test_require_existing_scroll_non_404_remains_transport_error():
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
fake.payload_indexes = set(_REQUIRED_INDEXES)
original_request = fake.request
def request(method, url, **kwargs):
if method == "POST" and url.endswith("/points/scroll"):
original_request(method, url, **kwargs)
return FakeResponse(503, {"status": "error"})
return original_request(method, url, **kwargs)
store = QdrantVectorStore(
base_url="http://qdrant:6333", collection="workspace-semantic", workspace_id="demo",
workspace_revision="a" * 40, expected_dimension=1024,
collection_lifecycle="require_existing", request=request,
)
with pytest.raises(VectorTransportError) as caught:
store.existing_hashes("memory", ["memory"])
assert caught.value.status_code == 503
def test_require_existing_scroll_malformed_remains_response_error():
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
fake.payload_indexes = set(_REQUIRED_INDEXES)
fake.malformed_scroll = True
store = _store(fake, collection_lifecycle="require_existing")
with pytest.raises(VectorResponseError):
store.existing_hashes("memory", ["memory"])
@pytest.mark.parametrize(
("dimension", "distance", "indexes", "index_types"),
[(384, "Cosine", set(), {}),
(1024, "Dot", set(), {}),
(1024, "Cosine", {"content_hash"}, {}),
(1024, "Cosine", {
"content_hash", "document_id", "kind", "record_key", "record_kind",
"vector_generation", "workspace_id", "workspace_revision",
}, {"kind": "integer"})],
)
def test_require_existing_refuses_incompatible_collection_without_mutations(
dimension, distance, indexes, index_types
):
fake = FakeQdrantHttp(dimension=dimension, distance=distance)
fake.collection = {"vectors": {"size": dimension, "distance": distance}}
fake.payload_indexes = indexes
fake.payload_index_types = index_types
store = _store(fake, collection_lifecycle="require_existing")
with pytest.raises(VectorStoreError, match="semantic_index_incompatible"):
store.upsert("memory", [_write_record("memory:1", "memory")])
assert [call for call in fake.calls if call[0] == "PUT"] == []
def test_require_existing_writes_compatible_collection_without_lifecycle_mutations():
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
fake.payload_indexes = {
"content_hash", "document_id", "kind", "record_key", "record_kind",
"vector_generation", "workspace_id", "workspace_revision",
}
store = _store(fake, collection_lifecycle="require_existing")
assert store.upsert("memory", [_write_record("memory:1", "memory")]) == 1
assert not [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/index")]
def test_require_existing_write_fails_after_collection_is_deleted_without_recreating():
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
fake.payload_indexes = {
"content_hash", "document_id", "kind", "record_key", "record_kind",
"vector_generation", "workspace_id", "workspace_revision",
}
original_request = fake.request
deleted = False
def request(method, url, **kwargs):
nonlocal deleted
response = original_request(method, url, **kwargs)
if method == "GET" and url.endswith("/collections/workspace-semantic") and not deleted:
deleted = True
fake.collection = None
return response
store = QdrantVectorStore(
base_url="http://qdrant:6333", collection="workspace-semantic", workspace_id="demo",
workspace_revision="a" * 40, expected_dimension=1024,
collection_lifecycle="require_existing", request=request,
)
from tht.ports.vector import SemanticIndexIncompatibleError
with pytest.raises(SemanticIndexIncompatibleError) as caught:
store.upsert("memory", [_write_record("memory:1", "memory")])
assert caught.value.code == "semantic_index_incompatible"
assert not [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")]
def test_upsert_creates_collection_and_keyword_indexes_idempotently():
fake = FakeQdrantHttp()
store = _store(fake)
assert store.upsert("memory", [_write_record("memory:1", "memory")]) == 1
assert store.upsert("memory", [_write_record("memory:1", "memory")]) == 1
creates = [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")]
assert len(creates) == 1
assert creates[0][2] == {"vectors": {"size": 1024, "distance": "Cosine"}}
assert fake.payload_indexes == {
"content_hash",
"document_id",
"kind",
"record_key",
"record_kind",
"vector_generation",
"workspace_id",
"workspace_revision",
}
def test_upsert_refuses_collection_dimension_or_distance_mismatch_without_recreating():
fake = FakeQdrantHttp(dimension=384, distance="Dot")
fake.collection = {"vectors": {"size": 384, "distance": "Dot"}}
store = _store(fake)
with pytest.raises(VectorStoreError, match="Qdrant collection configuration mismatch"):
store.upsert("memory", [_write_record("memory:1", "memory")])
creates = [call for call in fake.calls if call[0] == "PUT" and call[1].endswith("/collections/workspace-semantic")]
assert creates == []
def test_health_fails_when_the_bound_collection_is_missing():
fake = FakeQdrantHttp()
health = _store(fake).health()
assert health.ok is False
assert health.read_reachable is False
assert health.write_reachable is False
assert "missing" in (health.detail or "").lower()
def test_health_fails_when_required_payload_indexes_are_missing_without_creating_them():
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
health = _store(fake).health()
assert health.ok is False
assert fake.payload_indexes == set()
@pytest.mark.parametrize(
("record", "semantic_kind"),
[
(_write_record("schema_column:patients.id", "schema_column"), "schema"),
(
_write_record(
"demo:gen:11111111111111111111111111111111:chunk:1",
"evidence",
metadata={
"workspace_id": "demo",
"vector_generation": "gen:11111111111111111111111111111111",
"document_id": "doc:abc",
},
),
"evidence",
),
(_write_record("memory:1", "memory"), "memory"),
],
)
def test_upsert_serializes_qdrant_point_payloads(record, semantic_kind):
fake = FakeQdrantHttp()
store = _store(fake)
store.upsert("memory" if semantic_kind == "memory" else "evidence" if semantic_kind == "evidence" else "schema_records", [record])
point = next(iter(fake.points.values()))
assert point["id"] == point_id("demo", semantic_kind, record.record.id)
assert point["vector"] == record.embedding
assert point["payload"]["workspace_id"] == "demo"
assert point["payload"]["workspace_revision"] == "a" * 40
assert point["payload"]["kind"] == semantic_kind
assert point["payload"]["record_kind"] == record.record.kind
assert point["payload"]["record_key"] == record.record.id
assert point["payload"]["content_hash"] == record.content_hash
def test_search_filters_by_workspace_and_allowed_record_kinds():
fake = FakeQdrantHttp()
store = _store(fake)
store.upsert("memory", [_write_record("memory:1", "memory")])
other = next(iter(fake.points.values())).copy()
other["id"] = point_id("other", "memory", "memory:2")
other["payload"] = {**other["payload"], "workspace_id": "other", "record_key": "memory:2"}
fake.points[other["id"]] = other
solved = next(iter(fake.points.values())).copy()
solved["id"] = point_id("demo", "memory", "solved:1")
solved["payload"] = {**solved["payload"], "record_key": "solved:1", "record_kind": "solved_question"}
fake.points[solved["id"]] = solved
hits = store.search(["memory"], [0.2] * 1024, limit=5, kinds=["memory"])
assert [hit.id for hit in hits] == ["memory:1"]
query_call = next(call for call in fake.calls if call[0] == "POST" and call[1].endswith("/points/query?wait=true") is False and call[1].endswith("/points/query"))
assert query_call[2]["filter"] == {
"must": [
{"key": "workspace_id", "match": {"value": "demo"}},
{"key": "kind", "match": {"any": ["memory"]}},
{"key": "record_kind", "match": {"any": ["memory"]}},
]
}
def test_search_excludes_inconsistent_semantic_kind_in_bound_workspace():
fake = FakeQdrantHttp()
store = _store(fake)
store.upsert("memory", [_write_record("memory:1", "memory")])
contaminated = next(iter(fake.points.values())).copy()
contaminated["id"] = point_id("demo", "evidence", "memory:contaminated")
contaminated["payload"] = {
**contaminated["payload"],
"kind": "evidence",
"record_key": "memory:contaminated",
}
fake.points[contaminated["id"]] = contaminated
hits = store.search(["memory"], [0.2] * 1024, limit=5, kinds=["memory"])
assert [hit.id for hit in hits] == ["memory:1"]
def test_existing_hashes_health_and_exact_generation_inventory_and_delete():
fake = FakeQdrantHttp()
store = _store(fake)
generation = "gen:" + "1" * 32
keep = "gen:" + "2" * 32
store.upsert("evidence", [
_write_record(
f"demo:{generation}:chunk:1",
"evidence",
metadata={"workspace_id": "demo", "vector_generation": generation, "document_id": "doc:1"},
),
_write_record(
f"demo:{keep}:chunk:2",
"evidence",
metadata={"workspace_id": "demo", "vector_generation": keep, "document_id": "doc:2"},
),
])
assert store.existing_hashes("evidence", ["evidence"]) == {
f"demo:{generation}:chunk:1": "sha256:" + "a" * 64,
f"demo:{keep}:chunk:2": "sha256:" + "a" * 64,
}
assert store.list_evidence_generations("evidence", "demo") == [generation, keep]
assert store.delete_generation("evidence", generation, "demo") == 1
assert store.list_evidence_generations("evidence", "demo") == [keep]
health = store.health()
assert health.ok is True
assert health.read_reachable is True
assert health.write_reachable is True
assert health.observed_dimensions == (1024,)
assert health.dimension_compatible is True
def test_evidence_inventory_and_delete_ignore_inconsistent_semantic_kind():
fake = FakeQdrantHttp()
store = _store(fake)
generation = "gen:" + "1" * 32
contaminated_generation = "gen:" + "2" * 32
store.upsert("evidence", [
_write_record(
f"demo:{generation}:chunk:1",
"evidence",
metadata={
"workspace_id": "demo",
"vector_generation": generation,
"document_id": "doc:1",
},
),
])
contaminated_delete = next(iter(fake.points.values())).copy()
contaminated_delete["id"] = point_id("demo", "memory", "evidence:contaminated-delete")
contaminated_delete["payload"] = {
**contaminated_delete["payload"],
"kind": "memory",
"record_key": "evidence:contaminated-delete",
}
fake.points[contaminated_delete["id"]] = contaminated_delete
contaminated_list = next(iter(fake.points.values())).copy()
contaminated_list["id"] = point_id("demo", "memory", "evidence:contaminated-list")
contaminated_list["payload"] = {
**contaminated_list["payload"],
"kind": "memory",
"record_key": "evidence:contaminated-list",
"vector_generation": contaminated_generation,
}
fake.points[contaminated_list["id"]] = contaminated_list
assert store.list_evidence_generations("evidence", "demo") == [generation]
assert store.delete_generation("evidence", generation, "demo") == 1
assert contaminated_delete["id"] in fake.points
assert contaminated_list["id"] in fake.points
def test_metadata_search_rejects_a_workspace_id_different_from_the_bound_adapter():
fake = FakeQdrantHttp()
store = _store(fake)
generation = "gen:" + "1" * 32
store.upsert("evidence", [
_write_record(
f"demo:{generation}:chunk:1",
"evidence",
metadata={
"workspace_id": "demo", "vector_generation": generation,
"document_id": "doc:shared",
},
),
])
foreign = next(iter(fake.points.values())).copy()
foreign["id"] = point_id("other", "evidence", f"other:{generation}:chunk:1")
foreign["payload"] = {
**foreign["payload"],
"workspace_id": "other",
"record_key": f"other:{generation}:chunk:1",
"ref": "ref:foreign",
"title": "foreign",
"content": "foreign",
}
fake.points[foreign["id"]] = foreign
with pytest.raises(VectorStoreError, match="workspace namespace does not match"):
store.search(
["evidence"], [0.2] * 1024, limit=5, kinds=["evidence"],
metadata_filter={
"workspace_id": "other",
"vector_generation": generation,
"document_ids": ["doc:shared"],
},
)
def test_generation_inventory_rejects_a_workspace_id_different_from_the_bound_adapter():
fake = FakeQdrantHttp()
store = _store(fake)
with pytest.raises(VectorStoreError, match="workspace namespace does not match"):
store.list_evidence_generations("evidence", "other")
assert not any(call[1].endswith("/points/scroll") for call in fake.calls)
def test_generation_delete_cannot_mutate_foreign_workspace_or_non_evidence_points():
fake = FakeQdrantHttp()
store = _store(fake)
generation = "gen:" + "1" * 32
store.upsert("evidence", [
_write_record(
f"demo:{generation}:chunk:1",
"evidence",
metadata={
"workspace_id": "demo", "vector_generation": generation,
"document_id": "doc:demo",
},
),
])
demo = next(iter(fake.points.values()))
foreign = demo.copy()
foreign["id"] = point_id("other", "evidence", f"other:{generation}:chunk:1")
foreign["payload"] = {
**demo["payload"], "workspace_id": "other",
"record_key": f"other:{generation}:chunk:1",
}
fake.points[foreign["id"]] = foreign
memory = demo.copy()
memory["id"] = point_id("other", "memory", "memory:foreign")
memory["payload"] = {
**demo["payload"], "workspace_id": "other", "kind": "memory",
"record_kind": "memory", "record_key": "memory:foreign",
}
fake.points[memory["id"]] = memory
before = set(fake.points)
with pytest.raises(VectorStoreError, match="workspace namespace does not match"):
store.delete_generation("evidence", generation, "other")
assert set(fake.points) == before
assert not any(call[1].endswith("/points/delete?wait=true") for call in fake.calls)
def test_delete_kinds_is_workspace_scoped_and_preserves_other_semantic_kinds():
fake = FakeQdrantHttp()
store = _store(fake)
store.upsert("memory", [_write_record("memory:1", "memory")])
store.upsert("memory", [_write_record("solved:1", "solved_question")])
store.upsert("schema_records", [_write_record("schema_table:patients", "schema_table")])
other_workspace_memory = next(
point for point in fake.points.values() if point["payload"]["record_key"] == "memory:1"
).copy()
other_workspace_memory["id"] = point_id("other", "memory", "memory:other")
other_workspace_memory["payload"] = {
**other_workspace_memory["payload"],
"workspace_id": "other",
"record_key": "memory:other",
"title": "title:memory:other",
"content": "content:memory:other",
"ref": "ref:memory:other",
}
fake.points[other_workspace_memory["id"]] = other_workspace_memory
assert store.delete_kinds("memory", ["memory"]) == 1
delete_call = next(
call
for call in fake.calls
if call[0] == "POST" and call[1].endswith("/points/delete?wait=true")
)
assert delete_call[2]["filter"] == {
"must": [
{"key": "workspace_id", "match": {"value": "demo"}},
{"key": "kind", "match": {"any": ["memory"]}},
{"key": "record_kind", "match": {"any": ["memory"]}},
]
}
assert {
point["payload"]["record_key"]: point["payload"]["record_kind"]
for point in fake.points.values()
} == {
"solved:1": "solved_question",
"schema_table:patients": "schema_table",
"memory:other": "memory",
}
def test_existing_hashes_and_delete_kinds_ignore_inconsistent_semantic_kind():
fake = FakeQdrantHttp()
store = _store(fake)
store.upsert("memory", [_write_record("memory:1", "memory")])
contaminated = next(iter(fake.points.values())).copy()
contaminated["id"] = point_id("demo", "evidence", "memory:contaminated")
contaminated["payload"] = {
**contaminated["payload"],
"kind": "evidence",
"record_key": "memory:contaminated",
}
fake.points[contaminated["id"]] = contaminated
assert store.existing_hashes("memory", ["memory"]) == {
"memory:1": "sha256:" + "a" * 64,
}
assert store.delete_kinds("memory", ["memory"]) == 1
assert contaminated["id"] in fake.points
def test_sanitizes_timeout_and_malformed_responses():
fake = FakeQdrantHttp()
store = _store(fake)
fake.fail_request = requests.Timeout("dial tcp 10.0.0.9:6333: i/o timeout")
with pytest.raises(VectorStoreError, match="Qdrant request failed") as timeout:
store.search(["memory"], [0.2] * 1024, limit=1)
assert "10.0.0.9" not in str(timeout.value)
fake.fail_request = None
store.upsert("memory", [_write_record("memory:1", "memory")])
fake.malformed_query = True
with pytest.raises(VectorStoreError, match="Qdrant returned malformed query response"):
store.search(["memory"], [0.2] * 1024, limit=1)
fake.malformed_query = False
fake.malformed_scroll = True
with pytest.raises(VectorStoreError, match="Qdrant returned malformed scroll response"):
store.existing_hashes("memory", ["memory"])
def test_upsert_payload_keeps_canonical_identity_when_metadata_collides():
fake = FakeQdrantHttp()
store = _store(fake)
record = _write_record(
"memory:1",
"memory",
metadata={
"workspace_id": "evil",
"kind": "evil",
"record_kind": "evil",
"record_key": "evil",
"content_hash": "evil",
},
)
store.upsert("memory", [record])
payload = next(iter(fake.points.values()))["payload"]
assert payload["workspace_id"] == "demo"
assert payload["kind"] == "memory"
assert payload["record_kind"] == "memory"
assert payload["record_key"] == "memory:1"
assert payload["content_hash"] == "sha256:" + "a" * 64
def test_scroll_based_operations_paginate_until_next_page_offset_is_absent():
fake = FakeQdrantHttp()
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
generation_a = "gen:" + "1" * 32
generation_b = "gen:" + "2" * 32
fake.scroll_pages = [
{
"offset": None,
"points": [
{
"id": "p1",
"payload": {
"workspace_id": "demo",
"kind": "evidence",
"record_kind": "evidence",
"record_key": f"demo:{generation_a}:chunk:1",
"content_hash": "sha256:" + "a" * 64,
"vector_generation": generation_a,
},
}
],
"next_page_offset": "page-2",
},
{
"offset": "page-2",
"points": [
{
"id": "p2",
"payload": {
"workspace_id": "demo",
"kind": "evidence",
"record_kind": "evidence",
"record_key": f"demo:{generation_a}:chunk:2",
"content_hash": "sha256:" + "b" * 64,
"vector_generation": generation_a,
},
},
{
"id": "p3",
"payload": {
"workspace_id": "demo",
"kind": "evidence",
"record_kind": "evidence",
"record_key": f"demo:{generation_b}:chunk:3",
"content_hash": "sha256:" + "c" * 64,
"vector_generation": generation_b,
},
},
],
"next_page_offset": None,
},
]
store = _store(fake)
assert store.existing_hashes("evidence", ["evidence"]) == {
f"demo:{generation_a}:chunk:1": "sha256:" + "a" * 64,
f"demo:{generation_a}:chunk:2": "sha256:" + "b" * 64,
f"demo:{generation_b}:chunk:3": "sha256:" + "c" * 64,
}
assert store.list_evidence_generations("evidence", "demo") == [generation_a, generation_b]
assert store.delete_generation("evidence", generation_a, "demo") == 2
offsets = [
call[2].get("offset")
for call in fake.calls
if call[0] == "POST" and call[1].endswith("/points/scroll")
]
assert offsets[:2] == [None, "page-2"]