From 73bdac0da20e3140c072b803e66a1128a5069a9a Mon Sep 17 00:00:00 2001 From: Ali Zarinkolah Date: Sat, 22 Aug 2026 17:12:56 +0330 Subject: [PATCH] test(points): cover soft delete, relinking, and the file sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: - The failure modes worth testing are races, and they are cheap to force against the fake and expensive to observe anywhere else. Changes: - Unit tests for both boundaries, the repeat delete, a missing neighbour, the traversal property after several deletes, and — via a repository that bumps a rival's version before each apply — both the partial-apply repair and the unconvergent 409. - One new shared contract scenario (a multi-point batch applies every patch, including nulling a pointer) so it runs against the fake and real Qdrant. - HTTP tests against real Postgres and Qdrant for relinking, soft-not-hard delete, cross-tenant 404, and scope enforcement on both routes. - create_source_file factory for tests addressing a file without uploading. Co-Authored-By: Claude Opus 5 --- .../qdrant/test_points_deletion_api.py | 276 +++++++++++++ tests/support/factories.py | 31 ++ tests/support/point_contract.py | 46 +++ .../unit/application/points/test_deletion.py | 379 ++++++++++++++++++ 4 files changed, 732 insertions(+) create mode 100644 tests/integration/qdrant/test_points_deletion_api.py create mode 100644 tests/unit/application/points/test_deletion.py diff --git a/tests/integration/qdrant/test_points_deletion_api.py b/tests/integration/qdrant/test_points_deletion_api.py new file mode 100644 index 0000000..d6c6c84 --- /dev/null +++ b/tests/integration/qdrant/test_points_deletion_api.py @@ -0,0 +1,276 @@ +"""Soft delete over HTTP, against real Postgres and real Qdrant (ADR-0002). + +The unit tests in `tests/unit/application/points/test_deletion.py` decide +whether the relink logic is right. This file decides whether it is right +*through the stack*: real routing and scope checks, a tenant derived from a real +API key, and a real filtered `set_payload` batch — the last of which is the part +a fake can only approximate, since Qdrant reports success for a patch that +matched nothing. + +`DELETE /v1/files/{file_id}` is here too rather than with the upload tests: it +spans both stores, and the assertion that matters is the one about points. +""" + +import uuid +from collections.abc import AsyncIterator + +import pytest +import pytest_asyncio +from asgi_lifespan import LifespanManager +from httpx import ASGITransport, AsyncClient +from qdrant_client import AsyncQdrantClient +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.bootstrap.dependencies import get_sessionmaker +from src.config import QdrantSettings, Settings +from src.infrastructure.postgres.repositories import source_files as source_files_repo +from src.infrastructure.qdrant.collection import ensure_chunks_collection +from src.infrastructure.qdrant.points import QdrantPointStorage +from src.main import create_app +from tests.support.factories import create_api_key, create_source_file, create_tenant +from tests.support.point_contract import SeedSpec, chunk_point_for, point_id_for + +pytestmark = [ + pytest.mark.integration, + pytest.mark.qdrant, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + +FILE_ID = uuid.UUID("cccccccc-cccc-4ccc-8ccc-cccccccccccc") + + +@pytest.fixture +def api_settings(settings: Settings, qdrant_settings: QdrantSettings) -> Settings: + return settings.model_copy(update={"qdrant": qdrant_settings}) + + +@pytest_asyncio.fixture(loop_scope="session") +async def api_client( + api_settings: Settings, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> AsyncIterator[AsyncClient]: + app = create_app(api_settings) + app.dependency_overrides[get_sessionmaker] = lambda: db_sessionmaker + async with ( + LifespanManager(app) as manager, + AsyncClient(transport=ASGITransport(app=manager.app), base_url="http://test") as client, + ): + yield client + + +def _auth(token: str) -> dict[str, str]: + return {"Authorization": f"Bearer {token}"} + + +async def _tenant_with_key( + db_session: AsyncSession, *, scopes: list[str] | None = None +) -> tuple[uuid.UUID, str]: + tenant = await create_tenant(db_session) + _, token = await create_api_key( + db_session, + tenant=tenant, + scopes=scopes if scopes is not None else ["points:read", "points:write"], + ) + await db_session.commit() + return tenant.id, token + + +async def _seed_chain( + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, + *, + tenant_id: uuid.UUID, + file_id: uuid.UUID = FILE_ID, + length: int = 3, +) -> list[uuid.UUID]: + """A linked run of points, written the way ingestion writes them.""" + await ensure_chunks_collection(qdrant_client, collection=qdrant_settings.collection) + ids = [point_id_for(file_id, index) for index in range(length)] + points = [] + for index in range(length): + chunk_point = chunk_point_for( + SeedSpec(tenant_id, file_id, index, f"chunk {index}", float(index + 1)) + ) + chunk_point.payload["previous_chunk_id"] = str(ids[index - 1]) if index else None + chunk_point.payload["next_chunk_id"] = str(ids[index + 1]) if index + 1 < length else None + points.append(chunk_point) + + storage = QdrantPointStorage(qdrant_client, collection=qdrant_settings.collection) + await storage.upsert_points(points) + return ids + + +async def _read_point( + api_client: AsyncClient, token: str, point_id: uuid.UUID +) -> dict[str, object]: + response = await api_client.get(f"/v1/points/{point_id}", headers=_auth(token)) + assert response.status_code == 200 + payload: dict[str, object] = response.json() + return payload + + +async def test_delete_point_deactivates_it_and_relinks_its_neighbours( + api_client: AsyncClient, + db_session: AsyncSession, + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, +) -> None: + tenant_id, token = await _tenant_with_key(db_session) + first, middle, last = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id) + + response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token)) + + assert response.status_code == 200 + assert response.json()["is_active"] is False + assert response.json()["deleted_at"] is not None + assert (await _read_point(api_client, token, first))["next_chunk_id"] == str(last) + assert (await _read_point(api_client, token, last))["previous_chunk_id"] == str(first) + + +async def test_delete_point_keeps_the_point_in_qdrant( + api_client: AsyncClient, + db_session: AsyncSession, + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, +) -> None: + """Soft delete means soft: the point is still there, just not listed.""" + tenant_id, token = await _tenant_with_key(db_session) + _, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id) + + await api_client.delete(f"/v1/points/{middle}", headers=_auth(token)) + + listed = await api_client.get( + "/v1/points", params={"file_id": str(FILE_ID)}, headers=_auth(token) + ) + assert str(middle) not in [point["point_id"] for point in listed.json()["points"]] + assert (await _read_point(api_client, token, middle))["is_active"] is False + + +async def test_delete_point_is_a_noop_success_the_second_time( + api_client: AsyncClient, + db_session: AsyncSession, + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, +) -> None: + tenant_id, token = await _tenant_with_key(db_session) + first, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id) + await api_client.delete(f"/v1/points/{middle}", headers=_auth(token)) + version_after_first = (await _read_point(api_client, token, first))["version"] + + again = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token)) + + assert again.status_code == 200 + assert again.json()["is_active"] is False + assert (await _read_point(api_client, token, first))["version"] == version_after_first + + +async def test_delete_point_returns_404_for_another_tenants_point( + api_client: AsyncClient, + db_session: AsyncSession, + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, +) -> None: + """`404`, not `403`, and nothing is deactivated on the way to saying so.""" + owner_id, owner_token = await _tenant_with_key(db_session) + _, intruder_token = await _tenant_with_key(db_session) + _, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=owner_id) + + response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(intruder_token)) + + assert response.status_code == 404 + assert response.json()["error"]["code"] == "not_found" + assert (await _read_point(api_client, owner_token, middle))["is_active"] is True + + +async def test_delete_point_requires_the_points_write_scope( + api_client: AsyncClient, + db_session: AsyncSession, + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, +) -> None: + """A read-only key can see a point but must not be able to remove it.""" + tenant_id, token = await _tenant_with_key(db_session, scopes=["points:read"]) + _, middle, _ = await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant_id) + + response = await api_client.delete(f"/v1/points/{middle}", headers=_auth(token)) + + assert response.status_code == 403 + assert (await _read_point(api_client, token, middle))["is_active"] is True + + +async def test_delete_file_deactivates_every_point_and_retires_the_row( + api_client: AsyncClient, + db_session: AsyncSession, + db_sessionmaker: async_sessionmaker[AsyncSession], + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, +) -> None: + tenant = await create_tenant(db_session) + _, token = await create_api_key( + db_session, tenant=tenant, scopes=["points:read", "points:write"] + ) + file_id = uuid.uuid4() + await create_source_file(db_session, tenant=tenant, source_file_id=file_id) + await db_session.commit() + await _seed_chain(qdrant_client, qdrant_settings, tenant_id=tenant.id, file_id=file_id) + + response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(token)) + + assert response.status_code == 200 + assert response.json()["points_soft_deleted"] == 3 + listed = await api_client.get( + "/v1/points", params={"file_id": str(file_id)}, headers=_auth(token) + ) + assert listed.json()["points"] == [] + + async with db_sessionmaker() as session: + row = await source_files_repo.get_by_id( + session, tenant_id=tenant.id, source_file_id=file_id + ) + assert row is not None + assert row.status == "soft_deleted" + assert row.deleted_at is not None + + +async def test_delete_file_returns_404_for_another_tenants_file( + api_client: AsyncClient, + db_session: AsyncSession, + qdrant_client: AsyncQdrantClient, + qdrant_settings: QdrantSettings, +) -> None: + """The authorization check runs before the sweep, so nothing is deactivated.""" + owner = await create_tenant(db_session) + _, owner_token = await create_api_key( + db_session, tenant=owner, scopes=["points:read", "points:write"] + ) + intruder = await create_tenant(db_session) + _, intruder_token = await create_api_key( + db_session, tenant=intruder, scopes=["points:read", "points:write"] + ) + file_id = uuid.uuid4() + await create_source_file(db_session, tenant=owner, source_file_id=file_id) + await db_session.commit() + await _seed_chain(qdrant_client, qdrant_settings, tenant_id=owner.id, file_id=file_id) + + response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(intruder_token)) + + assert response.status_code == 404 + assert response.json()["error"]["code"] == "not_found" + listed = await api_client.get( + "/v1/points", params={"file_id": str(file_id)}, headers=_auth(owner_token) + ) + assert len(listed.json()["points"]) == 3 + + +async def test_delete_file_requires_the_points_write_scope( + api_client: AsyncClient, db_session: AsyncSession +) -> None: + tenant = await create_tenant(db_session) + _, token = await create_api_key(db_session, tenant=tenant, scopes=["files:write"]) + file_id = uuid.uuid4() + await create_source_file(db_session, tenant=tenant, source_file_id=file_id) + await db_session.commit() + + response = await api_client.delete(f"/v1/files/{file_id}", headers=_auth(token)) + + assert response.status_code == 403 diff --git a/tests/support/factories.py b/tests/support/factories.py index 638781e..be1c07e 100644 --- a/tests/support/factories.py +++ b/tests/support/factories.py @@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.application.auth.keys import generate_api_key, hash_secret from src.infrastructure.postgres.models.api_key import ApiKey +from src.infrastructure.postgres.models.source_file import SourceFile from src.infrastructure.postgres.models.tenant import Tenant from src.infrastructure.postgres.models.tenant_domain import TenantDomain @@ -50,6 +51,36 @@ async def create_api_key( return api_key, full_key +async def create_source_file( + session: AsyncSession, + *, + tenant: Tenant, + source_file_id: uuid.UUID | None = None, + domain: str = "general", + status: str = "active", +) -> SourceFile: + """A `source_files` row for tests that address a file without uploading one. + + `DELETE /v1/files/{file_id}` authorizes against this row before touching a + single point, so a delete test needs it even though the interesting state + lives in Qdrant. + """ + source_file = SourceFile( + id=source_file_id or uuid.uuid4(), + tenant_id=tenant.id, + domain=domain, + source_filename="handbook.docx", + source_type="docx", + content_sha256="0" * 64, + byte_size=1024, + storage_uri="s3://bucket/key", + status=status, + ) + session.add(source_file) + await session.flush() + return source_file + + async def create_tenant_domain( session: AsyncSession, *, diff --git a/tests/support/point_contract.py b/tests/support/point_contract.py index b0e27d9..8e1730e 100644 --- a/tests/support/point_contract.py +++ b/tests/support/point_contract.py @@ -329,6 +329,51 @@ async def scenario_patch_cannot_cross_tenants( assert victim.content == "onboarding checklist for new staff" +async def scenario_patch_batch_applies_every_patch_in_one_call( + repository: PointRepository, *, tenant_a: uuid.UUID, file_a: uuid.UUID +) -> None: + """One `apply_patches` call carrying several points applies all of them. + + This is the primitive soft delete is built on: a deactivation plus its two + neighbour relinks go out together, and a batch that silently applied only + its first operation would leave the pointer chain broken — the exact defect + ADR-0002 rules out. Nulling a pointer is included because `None` and + "absent" are different payload values, and only one of them clears a link. + """ + deleted, previous, following = (point_id_for(file_a, index) for index in (1, 0, 2)) + await repository.apply_patches( + tenant_id=tenant_a, + patches=[ + PayloadPatch( + point_id=deleted, + payload={"is_active": False, "version": 2}, + expected_version=1, + ), + PayloadPatch( + point_id=previous, + payload={"next_chunk_id": str(following), "version": 2}, + expected_version=1, + ), + PayloadPatch( + point_id=following, + payload={"previous_chunk_id": None, "version": 2}, + expected_version=1, + ), + ], + ) + + applied = { + point.point_id: point + for point in await repository.get_many( + tenant_id=tenant_a, point_ids=[deleted, previous, following] + ) + } + assert applied[deleted].is_active is False + assert applied[previous].next_chunk_id == following + assert applied[following].previous_chunk_id is None + assert [applied[point_id].version for point_id in (deleted, previous, following)] == [2, 2, 2] + + # Scenarios that only read, safe to run in any order against one seeded corpus. READ_SCENARIOS = ( scenario_get_returns_point_within_tenant, @@ -350,4 +395,5 @@ WRITE_SCENARIOS = ( scenario_patch_applies_when_version_matches, scenario_patch_is_a_noop_when_version_is_stale, scenario_patch_cannot_cross_tenants, + scenario_patch_batch_applies_every_patch_in_one_call, ) diff --git a/tests/unit/application/points/test_deletion.py b/tests/unit/application/points/test_deletion.py new file mode 100644 index 0000000..08dcac3 --- /dev/null +++ b/tests/unit/application/points/test_deletion.py @@ -0,0 +1,379 @@ +"""Soft delete and neighbour relinking against the fake repository (ADR-0016). + +The point of these tests is the pointer chain, not the HTTP surface. ADR-0002 +treats a partial relink as a defect, and the ways to produce one are all here: +deleting at either boundary, deleting the same point twice, losing a version +race half way through a batch, and pointing at a neighbour that is gone. + +`tests/integration/qdrant/test_points_deletion.py` runs the two central cases +against real Qdrant. What only this file can do cheaply is force the races. +""" + +import uuid +from collections.abc import Sequence +from dataclasses import dataclass +from typing import override + +import pytest +import structlog.testing + +from src.application.points.deletion import soft_delete_file_points, soft_delete_point +from src.application.points.errors import PointVersionConflictError +from src.application.points.point import Point, PointNotFoundError +from src.application.ports.point_repository import PayloadPatch +from tests.fakes import FakePointRepository +from tests.support.point_contract import SeedSpec, build_point + +pytestmark = [pytest.mark.unit, pytest.mark.asyncio] + +TENANT = uuid.UUID("11111111-1111-4111-8111-111111111111") +OTHER_TENANT = uuid.UUID("22222222-2222-4222-8222-222222222222") +FILE = uuid.UUID("33333333-3333-4333-8333-333333333333") +OTHER_FILE = uuid.UUID("44444444-4444-4444-8444-444444444444") +ACTOR = "api_key:test" + + +def _chain(count: int, *, tenant: uuid.UUID = TENANT, file_id: uuid.UUID = FILE) -> list[Point]: + """`count` points of one file, linked head to tail in `order_id` order.""" + points = [ + build_point(SeedSpec(tenant, file_id, index, f"chunk {index}", float(index + 1))) + for index in range(count) + ] + return [ + point.model_copy( + update={ + "previous_chunk_id": points[index - 1].point_id if index else None, + "next_chunk_id": (points[index + 1].point_id if index + 1 < len(points) else None), + } + ) + for index, point in enumerate(points) + ] + + +def _seeded(*points: Point) -> FakePointRepository: + repository = FakePointRepository() + for point in points: + repository.add(point) + return repository + + +def _stored(repository: FakePointRepository, point_id: uuid.UUID) -> Point: + return repository.points[str(point_id)] + + +def _walk(repository: FakePointRepository, head: uuid.UUID) -> list[uuid.UUID]: + """Follow `next_chunk_id` from `head`, guarding against a cycle.""" + visited: list[uuid.UUID] = [] + current: uuid.UUID | None = head + while current is not None and len(visited) <= len(repository.points): + visited.append(current) + current = _stored(repository, current).next_chunk_id + return visited + + +async def test_soft_delete_point_relinks_the_neighbours_of_a_middle_point() -> None: + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + + deleted = await soft_delete_point( + repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR + ) + + assert deleted.is_active is False + assert deleted.deleted_at is not None + assert _stored(repository, first.point_id).next_chunk_id == last.point_id + assert _stored(repository, last.point_id).previous_chunk_id == first.point_id + + +async def test_soft_delete_point_sends_the_deactivation_and_both_relinks_in_one_batch() -> None: + """A partial relink is a defect, so the three patches must not be split up.""" + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + + await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR) + + assert len(repository.patch_batches) == 1 + assert {patch.point_id for patch in repository.patch_batches[0]} == { + middle.point_id, + first.point_id, + last.point_id, + } + + +async def test_soft_delete_point_bumps_the_version_of_every_point_it_touches() -> None: + """A relinked neighbour really changed, so a stale editor of it must `409`.""" + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + + await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR) + + assert [_stored(repository, point.point_id).version for point in (first, middle, last)] == [ + 2, + 2, + 2, + ] + assert _stored(repository, first.point_id).updated_by == ACTOR + + +async def test_soft_delete_point_leaves_the_new_head_without_a_previous_pointer() -> None: + first, second, third = _chain(3) + repository = _seeded(first, second, third) + + await soft_delete_point(repository, tenant_id=TENANT, point_id=first.point_id, actor=ACTOR) + + assert _stored(repository, second.point_id).previous_chunk_id is None + + +async def test_soft_delete_point_leaves_the_new_tail_without_a_next_pointer() -> None: + first, second, third = _chain(3) + repository = _seeded(first, second, third) + + await soft_delete_point(repository, tenant_id=TENANT, point_id=third.point_id, actor=ACTOR) + + assert _stored(repository, second.point_id).next_chunk_id is None + + +async def test_soft_delete_point_keeps_the_deleted_points_own_pointers() -> None: + """Nothing active points at it any more, so its pointers record where it sat. + + That record is what the retry re-plans from, and what a later restore or an + audit reader would need to place the point back in the sequence. + """ + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + + await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR) + + stored = _stored(repository, middle.point_id) + assert stored.previous_chunk_id == first.point_id + assert stored.next_chunk_id == last.point_id + + +async def test_soft_delete_point_is_a_noop_for_an_already_inactive_point() -> None: + """Not a `404`, and not a second relink — no patch is issued at all.""" + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR) + versions_after_first = { + point.point_id: _stored(repository, point.point_id).version + for point in (first, middle, last) + } + repository.patch_batches.clear() + + again = await soft_delete_point( + repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR + ) + + assert again.is_active is False + assert repository.patch_batches == [] + assert { + point.point_id: _stored(repository, point.point_id).version + for point in (first, middle, last) + } == versions_after_first + + +async def test_soft_delete_point_raises_not_found_for_another_tenants_point() -> None: + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + + with pytest.raises(PointNotFoundError): + await soft_delete_point( + repository, tenant_id=OTHER_TENANT, point_id=middle.point_id, actor=ACTOR + ) + + assert repository.patch_batches == [] + assert _stored(repository, middle.point_id).is_active is True + + +async def test_soft_delete_point_raises_not_found_for_an_unknown_point() -> None: + repository = _seeded(*_chain(2)) + + with pytest.raises(PointNotFoundError): + await soft_delete_point(repository, tenant_id=TENANT, point_id=uuid.uuid4(), actor=ACTOR) + + +async def test_soft_delete_point_skips_a_neighbour_that_is_not_there() -> None: + """A pointer naming an absent point means the chain was already broken. + + The delete completes the half of the relink that exists rather than + refusing, which would leave the point unremovable through any endpoint. + """ + first, middle, last = _chain(3) + repository = _seeded(first, middle) # `last` is never stored + + deleted = await soft_delete_point( + repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR + ) + + assert deleted.is_active is False + assert _stored(repository, first.point_id).next_chunk_id == last.point_id + + +async def test_soft_delete_point_visits_every_active_point_exactly_once_after_several_deletes() -> ( + None +): + """The plan's traversal property, over a sequence of deletes. + + Walking `next_chunk_id` from the head must reach every surviving point once + and never step into a deactivated one. + """ + points = _chain(5) + repository = _seeded(*points) + + for index in (1, 3): + await soft_delete_point( + repository, tenant_id=TENANT, point_id=points[index].point_id, actor=ACTOR + ) + + walked = _walk(repository, points[0].point_id) + assert walked == [points[0].point_id, points[2].point_id, points[4].point_id] + assert all(_stored(repository, point_id).is_active for point_id in walked) + + +@dataclass +class _ContendedRepository(FakePointRepository): + """Bumps one point's version just before an apply, as a rival writer would. + + That makes the patch guarding on the old version match nothing while the + rest of the batch lands — Qdrant's real behaviour, and the partial apply the + service's retry exists to repair. `rounds` bounds how long the rival keeps + interfering. + """ + + rival: uuid.UUID | None = None + rounds: int = 0 + + @override + async def apply_patches(self, *, tenant_id: uuid.UUID, patches: Sequence[PayloadPatch]) -> None: + if self.rounds > 0 and self.rival is not None: + self.rounds -= 1 + victim = self.points[str(self.rival)] + self.points[str(self.rival)] = victim.model_copy(update={"version": victim.version + 1}) + await super().apply_patches(tenant_id=tenant_id, patches=patches) + + +async def test_soft_delete_point_repairs_a_partially_applied_batch_on_retry() -> None: + first, middle, last = _chain(3) + repository = _ContendedRepository(rival=first.point_id, rounds=1) + for point in (first, middle, last): + repository.add(point) + + await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR) + + # The first batch left `first` stale; the second re-planned only what was + # still missing, rather than re-deactivating the already-inactive point. + assert len(repository.patch_batches) == 2 + assert [patch.point_id for patch in repository.patch_batches[1]] == [first.point_id] + assert _stored(repository, first.point_id).next_chunk_id == last.point_id + assert _stored(repository, middle.point_id).is_active is False + + +async def test_soft_delete_point_conflicts_when_the_relink_never_settles() -> None: + first, middle, last = _chain(3) + repository = _ContendedRepository(rival=first.point_id, rounds=99) + for point in (first, middle, last): + repository.add(point) + + with pytest.raises(PointVersionConflictError): + await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR) + + +async def test_soft_delete_file_points_deactivates_every_active_point_of_the_file() -> None: + points = _chain(3) + other_file = _chain(2, file_id=OTHER_FILE) + other_tenant = _chain(1, tenant=OTHER_TENANT, file_id=uuid.uuid4()) + repository = _seeded(*points, *other_file, *other_tenant) + + swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR) + + assert swept == 3 + assert all(not _stored(repository, point.point_id).is_active for point in points) + assert all(_stored(repository, point.point_id).is_active for point in other_file) + assert all(_stored(repository, point.point_id).is_active for point in other_tenant) + + +async def test_soft_delete_file_points_leaves_the_chain_intact() -> None: + """No survivor can dangle, so the sweep rewrites no pointer at all.""" + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + + await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR) + + assert _stored(repository, first.point_id).next_chunk_id == middle.point_id + assert _stored(repository, middle.point_id).previous_chunk_id == first.point_id + assert _stored(repository, last.point_id).previous_chunk_id == middle.point_id + + +async def test_soft_delete_file_points_pages_past_one_batch() -> None: + """More points than one sweep page, so the re-listing loop has to run.""" + points = _chain(230) + repository = _seeded(*points) + + swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR) + + assert swept == 230 + assert all(not _stored(repository, point.point_id).is_active for point in points) + + +async def test_soft_delete_file_points_is_a_noop_the_second_time() -> None: + repository = _seeded(*_chain(3)) + await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR) + repository.patch_batches.clear() + + swept = await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR) + + assert swept == 0 + assert repository.patch_batches == [] + + +async def test_soft_delete_file_points_returns_zero_for_an_unknown_file() -> None: + repository = _seeded(*_chain(2)) + + assert ( + await soft_delete_file_points( + repository, tenant_id=TENANT, file_id=uuid.uuid4(), actor=ACTOR + ) + == 0 + ) + + +async def test_soft_delete_point_logs_its_duration_and_round_count() -> None: + """ADR-0011's `duration_ms`, plus the field that explains a slow one. + + Relinking is O(1), so a delete's cost is Qdrant round trips; `rounds` above + 1 means a concurrent writer forced a re-plan rather than the store being + slow, and the two fields are only useful together. + """ + first, middle, last = _chain(3) + repository = _seeded(first, middle, last) + + with structlog.testing.capture_logs() as logs: + await soft_delete_point(repository, tenant_id=TENANT, point_id=middle.point_id, actor=ACTOR) + + event = next(entry for entry in logs if entry["event"] == "points.soft_deleted") + assert event["rounds"] == 1 + assert isinstance(event["duration_ms"], float) + + +async def test_soft_delete_file_points_logs_a_round_per_sweep_page() -> None: + """The sweep is the delete path whose cost tracks the size of the file.""" + repository = _seeded(*_chain(230)) + + with structlog.testing.capture_logs() as logs: + await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR) + + event = next(entry for entry in logs if entry["event"] == "points.file_soft_deleted") + assert event["points_soft_deleted"] == 230 + assert event["rounds"] == 3 + assert isinstance(event["duration_ms"], float) + + +async def test_soft_delete_file_points_conflicts_when_a_round_makes_no_progress() -> None: + """A sweep round that attempts the same ids as the one before is stuck.""" + points = _chain(3) + repository = _ContendedRepository(rival=points[0].point_id, rounds=99) + for point in points: + repository.add(point) + + with pytest.raises(PointVersionConflictError): + await soft_delete_file_points(repository, tenant_id=TENANT, file_id=FILE, actor=ACTOR)