"""Hand-written fakes for narrow application-owned ports (ADR-0016).""" import asyncio import uuid from collections.abc import Sequence from dataclasses import dataclass, field from datetime import datetime from src.application.ingestion.models import SparseVector from src.application.points.models import ChunkPoint from src.application.points.point import Point from src.application.ports.point_repository import PayloadPatch, PointPage @dataclass class FakeObjectStorage: """In-memory `ObjectStorage`. `fail_next` simulates one upload failure.""" objects: dict[str, bytes] = field(default_factory=dict) fail_next: bool = False async def put_object(self, *, key: str, data: bytes, content_type: str) -> None: if self.fail_next: self.fail_next = False raise OSError("simulated object storage failure") self.objects[key] = data @dataclass class FakeDenseEmbedder: """A scripted `DenseEmbedder`. Returns a fixed-dimension zero vector per text by default; `fail_next` simulates one batch failure. """ name: str dimensions: int = 4 value: float = 0.0 """Component value of every returned vector. Non-zero where a real Qdrant has to score the result, since a zero vector has no direction to compare.""" model_version: str = "fake-dense-v1" calls: list[list[str]] = field(default_factory=list) fail_next: bool = False delay_seconds: float = 0.0 """Simulates a slow provider call, e.g. to exercise timeout handling.""" async def embed_batch(self, texts: Sequence[str]) -> list[list[float]]: self.calls.append(list(texts)) if self.delay_seconds: await asyncio.sleep(self.delay_seconds) if self.fail_next: self.fail_next = False raise RuntimeError("simulated embedder failure") return [[self.value] * self.dimensions for _ in texts] @dataclass class FakeSparseEmbedder: """A scripted `SparseEmbedder`. Returns an empty sparse vector per text.""" name: str = "sparse" model_version: str = "fake-sparse-v1" calls: list[list[str]] = field(default_factory=list) fail_next: bool = False def embed_batch(self, texts: Sequence[str], *, query: bool = False) -> list[SparseVector]: self.calls.append(list(texts)) if self.fail_next: self.fail_next = False raise RuntimeError("simulated embedder failure") return [SparseVector(indices=[], values=[]) for _ in texts] @dataclass class FakePointStorage: """An in-memory `PointStorage`. `points` is keyed by point id, so a re-upsert of the same deterministic id overwrites rather than accumulating — the property a test asserting "a retry produces no duplicate points" needs the fake to actually model. `fail_on_batch` fails the Nth (0-based) upsert batch, which is how a test checks that the soft-delete sweep never runs after a partial failure. """ points: dict[str, ChunkPoint] = field(default_factory=dict) upsert_batches: list[int] = field(default_factory=list) deactivate_calls: list[dict[str, object]] = field(default_factory=list) fail_on_batch: int | None = None fail_deactivate: bool = False max_in_flight: int = 0 _in_flight: int = 0 async def upsert_points(self, points: Sequence[ChunkPoint]) -> None: self._in_flight += 1 self.max_in_flight = max(self.max_in_flight, self._in_flight) try: # Yield so concurrent batches actually overlap; without this the # in-flight ceiling is trivially 1 and the bound goes untested. await asyncio.sleep(0) index = len(self.upsert_batches) self.upsert_batches.append(len(points)) if self.fail_on_batch is not None and index == self.fail_on_batch: raise RuntimeError("simulated point storage failure") for point in points: self.points[str(point.point_id)] = point finally: self._in_flight -= 1 async def deactivate_points_from_index( self, *, tenant_id: uuid.UUID, file_id: uuid.UUID, from_chunk_index: int, deleted_at: datetime, updated_by: str, ) -> int: self.deactivate_calls.append( { "tenant_id": tenant_id, "file_id": file_id, "from_chunk_index": from_chunk_index, "updated_by": updated_by, } ) if self.fail_deactivate: raise RuntimeError("simulated deactivate failure") def is_stale(point: ChunkPoint) -> bool: chunk_index = point.payload.get("chunk_index") return ( point.payload.get("file_id") == str(file_id) and point.payload.get("tenant_id") == str(tenant_id) and point.payload.get("is_active") is True and isinstance(chunk_index, int) and chunk_index >= from_chunk_index ) stale = [point for point in self.points.values() if is_stale(point)] for point in stale: point.payload["is_active"] = False point.payload["deleted_at"] = deleted_at.isoformat() return len(stale) @dataclass class FakePointRepository: """An in-memory `PointRepository`. Stores `Point` models keyed by point id. The filter, ordering, and version semantics below are reimplemented in Python rather than stubbed, because every service this fake stands behind is being tested *for* those semantics: a stub that returned points in insertion order would pass a reorder test that real Qdrant fails. Three behaviours are modelled deliberately: - `is_active` is filtered out unless `include_inactive`, matching ADR-0002's implied read filter. - `list_by_file` sorts by `order_id` and paginates by value, the same cursor mechanism the Qdrant adapter uses, so a test can catch a cursor that skips or repeats. - `apply_patches` honours `expected_version` by silently matching nothing on a mismatch — the no-op-not-error behaviour Qdrant's filtered `set_payload` actually has, which is what makes the service's post-check necessary. """ points: dict[str, Point] = field(default_factory=dict) patch_batches: list[list[PayloadPatch]] = field(default_factory=list) fail_next: bool = False def add(self, point: Point) -> None: """Seed a point. Test-only helper, not part of the port.""" self.points[str(point.point_id)] = point def _visible( self, *, tenant_id: uuid.UUID, domain: str | None = None, file_id: uuid.UUID | None = None, include_inactive: bool = False, ) -> list[Point]: return [ point for point in self.points.values() if point.tenant_id == tenant_id and (domain is None or point.domain == domain) and (file_id is None or point.file_id == file_id) and (include_inactive or point.is_active) ] async def get( self, *, tenant_id: uuid.UUID, point_id: uuid.UUID, with_vectors: bool = False ) -> Point | None: point = self.points.get(str(point_id)) if point is None or point.tenant_id != tenant_id: return None return point if with_vectors else point.model_copy(update={"vectors": None}) async def get_many( self, *, tenant_id: uuid.UUID, point_ids: Sequence[uuid.UUID] ) -> tuple[Point, ...]: wanted = {str(point_id) for point_id in point_ids} return tuple( point for key, point in self.points.items() if key in wanted and point.tenant_id == tenant_id ) async def list_by_file( self, *, tenant_id: uuid.UUID, file_id: uuid.UUID, limit: int, cursor: str | None = None, include_inactive: bool = False, ) -> PointPage: ordered = sorted( self._visible(tenant_id=tenant_id, file_id=file_id, include_inactive=include_inactive), key=lambda point: point.order_id, ) if cursor is not None: ordered = [point for point in ordered if point.order_id > float(cursor)] page = tuple(ordered[:limit]) next_cursor = repr(page[-1].order_id) if len(page) == limit else None return PointPage(points=page, next_cursor=next_cursor) async def count( self, *, tenant_id: uuid.UUID, domain: str | None = None, file_id: uuid.UUID | None = None, include_inactive: bool = False, ) -> int: return len( self._visible( tenant_id=tenant_id, domain=domain, file_id=file_id, include_inactive=include_inactive, ) ) async def keyword_search( self, *, tenant_id: uuid.UUID, query: str, limit: int, cursor: str | None = None, domain: str | None = None, file_id: uuid.UUID | None = None, include_inactive: bool = False, ) -> PointPage: # Whole-token containment, lowercased: an approximation of Qdrant's # full-text index good enough to tell "matched" from "did not", which is # all the service layer branches on. Ranking is not modelled because the # real index does not rank either (ADR-0002). terms = query.lower().split() matches = [ point for point in self._visible( tenant_id=tenant_id, domain=domain, file_id=file_id, include_inactive=include_inactive, ) if all(term in point.content.lower().split() for term in terms) ] start = int(cursor) if cursor is not None else 0 page = tuple(matches[start : start + limit]) next_start = start + limit return PointPage( points=page, next_cursor=str(next_start) if next_start < len(matches) else None ) async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None: self.patch_batches.append(list(patches)) if self.fail_next: self.fail_next = False raise RuntimeError("simulated point repository failure") for patch in patches: point = self.points.get(str(patch.point_id)) if point is None or point.tenant_id != tenant_id: continue if patch.expected_version is not None and point.version != patch.expected_version: continue # Re-validated rather than `model_copy`d, because a patch payload # carries wire values (an ISO `deleted_at`, a stringified # `previous_chunk_id`) exactly as it would reach Qdrant. Copying # without validation would leave the fake holding a `str` where a # read from real Qdrant returns a `datetime`, and a service bug that # depends on the difference would pass here and fail in production. updated = point.model_dump() updated.update(patch.payload) self.points[str(patch.point_id)] = Point.model_validate(updated)