fix: harden qdrant response parsing and integration markers
This commit is contained in:
@@ -698,3 +698,197 @@ def test_scroll_based_operations_paginate_until_next_page_offset_is_absent():
|
||||
if call[0] == "POST" and call[1].endswith("/points/scroll")
|
||||
]
|
||||
assert offsets[:2] == [None, "page-2"]
|
||||
|
||||
|
||||
_MISSING = object()
|
||||
|
||||
|
||||
def _set_response_path(payload, path, value):
|
||||
if value is _MISSING:
|
||||
parent = payload
|
||||
for key in path[:-1]:
|
||||
parent = parent[key]
|
||||
parent.pop(path[-1], None)
|
||||
return
|
||||
parent = payload
|
||||
for key in path[:-1]:
|
||||
parent = parent[key]
|
||||
parent[path[-1]] = value
|
||||
|
||||
|
||||
def _collection_response_with_shape(fake, path, value):
|
||||
original = fake.request
|
||||
|
||||
def request(method, url, **kwargs):
|
||||
response = original(method, url, **kwargs)
|
||||
if method == "GET" and url.endswith("/collections/workspace-semantic") and response.ok:
|
||||
payload = response.json()
|
||||
_set_response_path(payload, path, value)
|
||||
return FakeResponse(200, payload)
|
||||
return response
|
||||
|
||||
return request
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("path", "value"),
|
||||
[
|
||||
(("result",), None),
|
||||
(("result",), []),
|
||||
(("result",), "result"),
|
||||
(("result",), _MISSING),
|
||||
(("result", "config"), None),
|
||||
(("result", "config"), []),
|
||||
(("result", "config"), "config"),
|
||||
(("result", "config"), _MISSING),
|
||||
(("result", "config", "params"), None),
|
||||
(("result", "config", "params"), []),
|
||||
(("result", "config", "params"), "params"),
|
||||
(("result", "config", "params"), _MISSING),
|
||||
(("result", "config", "params", "vectors"), None),
|
||||
(("result", "config", "params", "vectors"), []),
|
||||
(("result", "config", "params", "vectors"), "vectors"),
|
||||
(("result", "config", "params", "vectors"), _MISSING),
|
||||
(("result", "config", "params", "vectors", "size"), None),
|
||||
(("result", "config", "params", "vectors", "size"), []),
|
||||
(("result", "config", "params", "vectors", "size"), "1024"),
|
||||
(("result", "config", "params", "vectors", "size"), _MISSING),
|
||||
(("result", "config", "params", "vectors", "distance"), None),
|
||||
(("result", "config", "params", "vectors", "distance"), []),
|
||||
(("result", "config", "params", "vectors", "distance"), 1),
|
||||
(("result", "config", "params", "vectors", "distance"), _MISSING),
|
||||
(("result", "payload_schema"), None),
|
||||
(("result", "payload_schema"), []),
|
||||
(("result", "payload_schema"), "schema"),
|
||||
(("result", "payload_schema"), _MISSING),
|
||||
(("result", "payload_schema", "kind"), None),
|
||||
(("result", "payload_schema", "kind"), []),
|
||||
(("result", "payload_schema", "kind"), "keyword"),
|
||||
(("result", "payload_schema", "kind", "data_type"), None),
|
||||
(("result", "payload_schema", "kind", "data_type"), []),
|
||||
(("result", "payload_schema", "kind", "data_type"), _MISSING),
|
||||
],
|
||||
)
|
||||
def test_collection_success_response_shapes_are_typed_errors(path, value):
|
||||
fake = FakeQdrantHttp()
|
||||
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
|
||||
fake.payload_indexes = set(_REQUIRED_INDEXES)
|
||||
request = _collection_response_with_shape(fake, path, value)
|
||||
store = QdrantVectorStore(
|
||||
base_url="http://qdrant:6333", collection="workspace-semantic", workspace_id="demo",
|
||||
expected_dimension=1024, request=request,
|
||||
)
|
||||
|
||||
with pytest.raises(VectorResponseError):
|
||||
store.upsert("memory", [_write_record("memory:1", "memory")])
|
||||
|
||||
health = store.health()
|
||||
assert health.ok is False
|
||||
assert health.read_reachable is False
|
||||
assert health.write_reachable is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload_value", [None, [], {}])
|
||||
def test_query_success_response_payload_leaf_shapes_are_typed_errors(payload_value):
|
||||
fake = FakeQdrantHttp()
|
||||
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
|
||||
fake.payload_indexes = set(_REQUIRED_INDEXES)
|
||||
original = fake.request
|
||||
|
||||
def request(method, url, **kwargs):
|
||||
response = original(method, url, **kwargs)
|
||||
if method == "POST" and url.endswith("/points/query") and response.ok:
|
||||
payload = response.json()
|
||||
payload["result"]["points"] = [{
|
||||
"id": "p1", "score": 0.9,
|
||||
"payload": {"record_key": payload_value},
|
||||
}]
|
||||
return FakeResponse(200, payload)
|
||||
return response
|
||||
|
||||
store = _store(fake)
|
||||
store._request = request
|
||||
with pytest.raises(VectorResponseError):
|
||||
store.search(["memory"], [0.2] * 1024, limit=1, kinds=["memory"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("point", [None, [], "point"])
|
||||
def test_query_success_response_point_shapes_are_typed_errors(point):
|
||||
fake = FakeQdrantHttp()
|
||||
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
|
||||
fake.payload_indexes = set(_REQUIRED_INDEXES)
|
||||
original = fake.request
|
||||
|
||||
def request(method, url, **kwargs):
|
||||
response = original(method, url, **kwargs)
|
||||
if method == "POST" and url.endswith("/points/query") and response.ok:
|
||||
payload = response.json()
|
||||
payload["result"]["points"] = [point]
|
||||
return FakeResponse(200, payload)
|
||||
return response
|
||||
|
||||
store = _store(fake)
|
||||
store._request = request
|
||||
with pytest.raises(VectorResponseError):
|
||||
store.search(["memory"], [0.2] * 1024, limit=1, kinds=["memory"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("path", "value"),
|
||||
[
|
||||
(("result",), None),
|
||||
(("result",), []),
|
||||
(("result",), "result"),
|
||||
(("result",), _MISSING),
|
||||
(("result", "points"), None),
|
||||
(("result", "points"), {}),
|
||||
(("result", "points"), "points"),
|
||||
(("result", "points"), _MISSING),
|
||||
(("result", "next_page_offset"), []),
|
||||
(("result", "next_page_offset"), {}),
|
||||
(("result", "next_page_offset"), 1.5),
|
||||
(("result", "next_page_offset"), True),
|
||||
],
|
||||
)
|
||||
def test_scroll_success_response_shapes_are_typed_errors(path, value):
|
||||
fake = FakeQdrantHttp()
|
||||
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
|
||||
fake.payload_indexes = set(_REQUIRED_INDEXES)
|
||||
original = fake.request
|
||||
|
||||
def request(method, url, **kwargs):
|
||||
response = original(method, url, **kwargs)
|
||||
if method == "POST" and url.endswith("/points/scroll") and response.ok:
|
||||
payload = response.json()
|
||||
_set_response_path(payload, path, value)
|
||||
return FakeResponse(200, payload)
|
||||
return response
|
||||
|
||||
store = QdrantVectorStore(
|
||||
base_url="http://qdrant:6333", collection="workspace-semantic", workspace_id="demo",
|
||||
expected_dimension=1024, request=request,
|
||||
)
|
||||
with pytest.raises(VectorResponseError):
|
||||
store.existing_hashes("memory", ["memory"])
|
||||
|
||||
|
||||
@pytest.mark.parametrize("next_page_offset", [None, _MISSING])
|
||||
def test_scroll_accepts_null_or_missing_terminal_offset(next_page_offset):
|
||||
fake = FakeQdrantHttp()
|
||||
fake.collection = {"vectors": {"size": 1024, "distance": "Cosine"}}
|
||||
fake.payload_indexes = set(_REQUIRED_INDEXES)
|
||||
original = fake.request
|
||||
|
||||
def request(method, url, **kwargs):
|
||||
response = original(method, url, **kwargs)
|
||||
if method == "POST" and url.endswith("/points/scroll") and response.ok:
|
||||
payload = response.json()
|
||||
_set_response_path(payload, ("result", "next_page_offset"), next_page_offset)
|
||||
return FakeResponse(200, payload)
|
||||
return response
|
||||
|
||||
store = QdrantVectorStore(
|
||||
base_url="http://qdrant:6333", collection="workspace-semantic", workspace_id="demo",
|
||||
expected_dimension=1024, request=request,
|
||||
)
|
||||
assert store.existing_hashes("memory", ["memory"]) == {}
|
||||
|
||||
@@ -477,6 +477,7 @@ def test_registry_render_chain_produces_require_existing_qdrant_configs(
|
||||
assert not [call for call in fake.calls if call[0] == "PUT"]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_registry_rendered_session_and_maintenance_configs_bind_equally(tmp_path):
|
||||
session_path, maintenance_path = _render_registry_runtime_configs(tmp_path)
|
||||
|
||||
|
||||
@@ -110,12 +110,16 @@ class QdrantVectorStore:
|
||||
expected_dimension=self._expected_dimension,
|
||||
)
|
||||
|
||||
dimension = info["config"]["params"]["vectors"]["size"]
|
||||
result = self._require_mapping(info, "collection")
|
||||
config = self._require_mapping(result.get("config"), "collection")
|
||||
params = self._require_mapping(config.get("params"), "collection")
|
||||
vectors = self._require_mapping(params.get("vectors"), "collection")
|
||||
dimension = vectors["size"]
|
||||
dimensions = (dimension,)
|
||||
dimension_compatible = (
|
||||
None if self._expected_dimension is None else dimensions == (self._expected_dimension,)
|
||||
)
|
||||
observed_distance = info["config"]["params"]["vectors"].get("distance")
|
||||
observed_distance = vectors["distance"]
|
||||
distance_compatible = (
|
||||
None if self._expected_distance is None else observed_distance == self._expected_distance
|
||||
)
|
||||
@@ -179,9 +183,12 @@ class QdrantVectorStore:
|
||||
"filter": {"must": filter_must},
|
||||
},
|
||||
)
|
||||
points = response.get("result", {}).get("points")
|
||||
result = self._require_mapping(response.get("result"), "query")
|
||||
points = result.get("points")
|
||||
if not isinstance(points, list):
|
||||
raise VectorResponseError("Qdrant returned malformed query response")
|
||||
if not all(isinstance(point, dict) for point in points):
|
||||
raise VectorResponseError("Qdrant returned malformed query response")
|
||||
hits = [self._hit_from_point(point) for point in points]
|
||||
return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:limit]
|
||||
|
||||
@@ -380,12 +387,14 @@ class QdrantVectorStore:
|
||||
{"field_name": field_name, "field_schema": "keyword"},
|
||||
)
|
||||
response = self._call("GET", f"/collections/{self._collection}", None)
|
||||
result = response.get("result") if isinstance(response, dict) else None
|
||||
config = result.get("config", {}).get("params", {}).get("vectors") if isinstance(result, dict) else None
|
||||
if not isinstance(config, dict):
|
||||
result = self._require_mapping(response.get("result"), "collection")
|
||||
config = self._require_mapping(result.get("config"), "collection")
|
||||
params = self._require_mapping(config.get("params"), "collection")
|
||||
vectors = self._require_mapping(params.get("vectors"), "collection")
|
||||
size = vectors.get("size")
|
||||
distance = vectors.get("distance")
|
||||
if type(size) is not int or size <= 0 or not isinstance(distance, str):
|
||||
raise VectorResponseError("Qdrant returned malformed collection response")
|
||||
size = config.get("size")
|
||||
distance = config.get("distance")
|
||||
if (
|
||||
self._expected_dimension is not None and size != self._expected_dimension
|
||||
) or (
|
||||
@@ -399,6 +408,11 @@ class QdrantVectorStore:
|
||||
payload_schema = result.get("payload_schema")
|
||||
if not isinstance(payload_schema, dict):
|
||||
raise VectorResponseError("Qdrant returned malformed collection response")
|
||||
if any(
|
||||
not isinstance(field, dict) or not isinstance(field.get("data_type"), str)
|
||||
for field in payload_schema.values()
|
||||
):
|
||||
raise VectorResponseError("Qdrant returned malformed collection response")
|
||||
for field_name in _KEYWORD_INDEXES:
|
||||
field = payload_schema.get(field_name)
|
||||
if not isinstance(field, dict) or field.get("data_type") != "keyword":
|
||||
@@ -437,23 +451,40 @@ class QdrantVectorStore:
|
||||
"Qdrant collection disappeared during semantic index reconciliation"
|
||||
) from exc
|
||||
raise
|
||||
result = response.get("result", {})
|
||||
result = self._require_mapping(response.get("result"), "scroll")
|
||||
page = result.get("points")
|
||||
if not isinstance(page, list):
|
||||
if not isinstance(page, list) or not all(isinstance(point, dict) for point in page):
|
||||
raise VectorResponseError("Qdrant returned malformed scroll response")
|
||||
points.extend(page)
|
||||
next_page_offset = result.get("next_page_offset")
|
||||
if next_page_offset is None:
|
||||
return points
|
||||
if type(next_page_offset) not in (int, str):
|
||||
raise VectorResponseError("Qdrant returned malformed scroll response")
|
||||
if next_page_offset in seen_offsets:
|
||||
raise VectorResponseError("Qdrant returned malformed scroll response")
|
||||
seen_offsets.add(next_page_offset)
|
||||
offset = next_page_offset
|
||||
|
||||
@staticmethod
|
||||
def _require_mapping(value: object, operation: str) -> dict:
|
||||
if not isinstance(value, dict):
|
||||
raise VectorResponseError(f"Qdrant returned malformed {operation} response")
|
||||
return value
|
||||
|
||||
def _hit_from_point(self, point: dict) -> VectorHit:
|
||||
if not isinstance(point, dict):
|
||||
raise VectorResponseError("Qdrant returned malformed query response")
|
||||
payload = point.get("payload")
|
||||
score = point.get("score")
|
||||
if not isinstance(payload, dict) or not isinstance(score, (int, float)):
|
||||
if (
|
||||
not isinstance(payload, dict)
|
||||
or type(score) not in (int, float)
|
||||
or any(
|
||||
key in payload and not isinstance(payload[key], str)
|
||||
for key in ("record_key", "record_kind", "kind", "ref", "title", "content")
|
||||
)
|
||||
):
|
||||
raise VectorResponseError("Qdrant returned malformed query response")
|
||||
return hit_from_metadata(float(score), payload)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user