fix(vector-sync): address round-3 review — gate purge route, bound set, nits

- app.py: register /api/v1/vector-sync/purge only when vector_sync_enabled, so
  it returns 404 (not a 500 from get_qdrant_client) when sync is off
- scanner: bound _consent_backstop_done so a long-running multi-tenant process
  with user churn can't grow it without limit (clears on overflow)
- purge route: distinct 400 for a missing doc_types key; enforce the admin
  check even for an empty no-op request (destructive route)
- tests: missing-key 400, admin-gated empty no-op, non-admin empty 403

The _consent_narrowed_doc_types precondition is enforced by its non-Optional
frozenset[str] signature (ty rejects a None caller). The httpx.BasicAuth
SonarCloud hotspot matches the existing webhook routes (false positive,
credential from the app-password store) — left consistent for UI triage.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-16 01:21:01 +02:00
co-authored by Claude Opus 4.8
parent 24b8000a71
commit cef477b877
4 changed files with 64 additions and 11 deletions
+30 -1
View File
@@ -112,10 +112,25 @@ def test_forbidden_when_not_admin(mocker):
purge.assert_not_called()
def test_empty_doc_types_is_noop(mocker):
def test_missing_doc_types_key_returns_400(mocker):
_patch_token(mocker)
purge = _patch_purge(mocker)
client = TestClient(_build_app())
resp = client.post("/api/v1/vector-sync/purge", json={})
assert resp.status_code == 400
purge.assert_not_called()
def test_empty_doc_types_is_admin_gated_noop(mocker):
# An empty (no-op) request still requires admin — this is a destructive route.
_patch_token(mocker, "admin")
_patch_basic_auth(mocker, "admin")
_patch_outbound_client(mocker)
_patch_groups(mocker, ["admin"])
purge = _patch_purge(mocker)
client = TestClient(_build_app())
resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": []})
@@ -124,6 +139,20 @@ def test_empty_doc_types_is_noop(mocker):
purge.assert_not_called()
def test_empty_doc_types_forbidden_for_non_admin(mocker):
_patch_token(mocker, "bob")
_patch_basic_auth(mocker, "bob")
_patch_outbound_client(mocker)
_patch_groups(mocker, ["users"])
purge = _patch_purge(mocker)
client = TestClient(_build_app())
resp = client.post("/api/v1/vector-sync/purge", json={"doc_types": []})
assert resp.status_code == 403
purge.assert_not_called()
def test_admin_purge_happy_path(mocker):
_patch_token(mocker, "admin")
_patch_basic_auth(mocker, "admin")