curo/kite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(14.1-01): RED contract tests for cloud detail parity + force re-analyze

Browse Source 
- backend/tests/test_cloud_detail_parity.py: RED tests asserting GET
  /api/cloud/connections/{id}/items/{item_id}/detail returns extracted_text,
  analysis_status, semantic_index_status, topics, provider metadata, and
  excludes credentials_enc/object_key; owner gets 200, foreign user gets 404,
  admin gets 403 via get_regular_user; stale items retain prior data; detail
  endpoint must not call hydrate_and_cache_bytes (D-05, D-07, D-18, T-14.1-01/02)
- backend/tests/test_cloud_reanalyze_force.py: RED tests asserting force=True
  on AnalysisEnqueueRequest bypasses already_current for explicit re-analyze;
  default enqueue still skips unchanged indexed items; single-item retry with
  no active job creates a new one; force flag is owner-scoped and admin-blocked
  (D-11, D-12, ANALYZE-05, ANALYZE-06, ANALYZE-07, T-14.1-02)
- frontend/src/views/__tests__/CloudDetailParity.test.js: RED tests asserting
  cloud-file-detail named route exists in router; paired local+cloud route
  parity; DocumentView does not contain Re-classify (D-09 regression); route
  prerequisite for navigation (D-01, D-19)
- frontend/src/components/storage/__tests__/StorageBrowser.parity.test.js: RED
  tests asserting cloud rows render TopicBadge and analysis-status indicator in
  same slot as local rows; Analyze/Re-analyze/Retry action slot by state; no
  Re-classify copy in cloud rows (D-06, D-08, D-09, D-10)

All four files collect cleanly. 11 backend failures and 7 frontend failures are
tied exclusively to the missing detail endpoint, force flag, route, and Re-analyze
copy — not to fixture or infrastructure errors.
This commit is contained in:
curo1305
2026-06-26 21:47:44 +02:00
parent 37a7174623
commit a76854e003
4 changed files with 1724 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_cloud_detail_parity.py
+505
View File
@@ -0,0 +1,505 @@
"""
Phase 14.1 Plan 01 — RED contract tests for cloud detail endpoint parity.
Tests pin the contract for the cloud detail endpoint before implementation exists.
All tests will fail (404 or missing field assertions) until Plan 02 adds:
- GET /api/cloud/connections/{connection_id}/items/{item_id:path}/detail
- CloudItemDetailOut schema with extracted_text, analysis_status,
semantic_index_status, topics, provider, display_name, parent_ref/location,
modified_at, size, content_type, capabilities, unsupported_analysis_reason
Requirements: CLOUD-02, ANALYZE-01, CACHE-03, CACHE-05
Threats: T-14.1-01 (information disclosure), T-14.1-02 (privilege escalation)
Decisions: D-05, D-07, D-18, D-20
"""
from __future__ import annotations
import uuid as _uuid
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
pytestmark = pytest.mark.asyncio
from tests.conftest import _TEST_USER_AGENT
# ── Target endpoint path (Plan 02 implements this) ────────────────────────────
_DETAIL_PATH_TEMPLATE = "/api/cloud/connections/{connection_id}/items/{item_id}/detail"
# ── Shared helpers — reused from test_cloud_security.py pattern ──────────────
async def _create_user_and_token(session, role: str = "user"):
"""Create User + Quota + JWT. Mirrors the pattern from test_cloud_security.py."""
from db.models import User, Quota
from services.auth import hash_password, create_access_token
user_id = _uuid.uuid4()
user = User(
id=user_id,
handle=f"det_user_{user_id.hex[:8]}",
email=f"det_{user_id.hex[:8]}@example.com",
password_hash=hash_password("Testpassword123!"),
role=role,
is_active=True,
password_must_change=False,
)
quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
session.add(user)
session.add(quota)
await session.commit()
await session.refresh(user)
token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
return {
"user": user,
"token": token,
"headers": {
"Authorization": f"Bearer {token}",
"User-Agent": _TEST_USER_AGENT,
},
}
async def _create_cloud_connection(
session,
user_id,
provider: str = "google_drive",
name: str = "Test Drive",
status: str = "ACTIVE",
):
"""Create a CloudConnection owned by user_id."""
from db.models import CloudConnection
from storage.cloud_utils import encrypt_credentials
from config import settings
master_key = settings.cloud_creds_key.encode()
creds_enc = encrypt_credentials(
master_key,
str(user_id),
{"access_token": "tok", "refresh_token": "ref"},
)
conn = CloudConnection(
id=_uuid.uuid4(),
user_id=user_id,
provider=provider,
display_name=name,
credentials_enc=creds_enc,
status=status,
)
session.add(conn)
await session.commit()
return conn
async def _create_cloud_item(
session,
user_id,
connection_id,
*,
name: str = "report.pdf",
kind: str = "file",
analysis_status: str = "pending",
extracted_text: str | None = None,
etag: str = "etag-v1",
):
"""Create a CloudItem with optional analysis fields."""
from db.models import CloudItem
item = CloudItem(
id=_uuid.uuid4(),
user_id=user_id,
connection_id=connection_id,
provider_item_id=f"pitem-det-{_uuid.uuid4().hex[:8]}",
name=name,
kind=kind,
content_type="application/pdf",
provider_size=102400,
etag=etag,
analysis_status=analysis_status,
semantic_index_status="none" if analysis_status == "pending" else "indexed",
extracted_text=extracted_text,
)
session.add(item)
await session.commit()
return item
async def _create_analyzed_item_with_topics(
session,
user_id,
connection_id,
topic_names: list[str] | None = None,
):
"""Create a CloudItem in 'indexed' state with CloudItemTopic links."""
from db.models import CloudItem, CloudItemTopic, Topic
from sqlalchemy import select
item = CloudItem(
id=_uuid.uuid4(),
user_id=user_id,
connection_id=connection_id,
provider_item_id=f"pitem-analyzed-{_uuid.uuid4().hex[:8]}",
name="analyzed.pdf",
kind="file",
content_type="application/pdf",
provider_size=204800,
etag="etag-analyzed-v1",
analysis_status="indexed",
semantic_index_status="indexed",
extracted_text="Extracted text from the analyzed document.",
)
session.add(item)
await session.flush()
topic_names = topic_names or ["finance", "contracts"]
for tname in topic_names:
# Find or create the topic
result = await session.execute(
select(Topic).where(Topic.name == tname)
)
topic = result.scalars().first()
if topic is None:
topic = Topic(id=_uuid.uuid4(), name=tname)
session.add(topic)
await session.flush()
link = CloudItemTopic(
cloud_item_id=item.id,
topic_id=topic.id,
)
session.add(link)
await session.commit()
return item
# ── T-14.1-01: Owner GET returns analysis fields ──────────────────────────────
async def test_owner_detail_returns_extracted_text_and_topics(async_client, db_session):
"""CLOUD-02 / D-05: Owner GET of the cloud detail endpoint returns analysis fields.
After analysis completes, the detail response must include:
extracted_text, analysis_status, semantic_index_status, topics (list of names),
provider, display_name, parent_ref, modified_at, size, content_type,
capabilities, and unsupported_analysis_reason.
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_analyzed_item_with_topics(
db_session, auth["user"].id, conn.id, topic_names=["finance", "contracts"]
)
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
resp = await async_client.get(url, headers=auth["headers"])
# Plan 02 implements this endpoint — currently returns 404 or 405
assert resp.status_code == 200, (
f"Expected 200 from cloud detail endpoint — "
f"got {resp.status_code}: {resp.text}. "
"Plan 02 must add GET /connections/{id}/items/{item_id}/detail."
)
body = resp.json()
# Analysis fields that must be present (D-05)
assert "extracted_text" in body, "Cloud detail must include extracted_text"
assert body["extracted_text"], "extracted_text must be non-empty for indexed item"
assert "analysis_status" in body, "Cloud detail must include analysis_status"
assert body["analysis_status"] == "indexed"
assert "semantic_index_status" in body, "Cloud detail must include semantic_index_status"
# Topics as a list of names
assert "topics" in body, "Cloud detail must include topics list"
topic_names = body["topics"]
assert isinstance(topic_names, list), "topics must be a list"
assert len(topic_names) >= 1, "topics must contain at least one name"
assert "finance" in topic_names or "contracts" in topic_names, (
f"Expected topic names in {topic_names}"
)
# Source metadata fields (D-04)
assert "provider" in body, "Cloud detail must include provider"
assert "display_name" in body, "Cloud detail must include display_name"
assert "size" in body, "Cloud detail must include size"
assert "content_type" in body, "Cloud detail must include content_type"
async def test_owner_detail_returns_capabilities_and_unsupported_reason(async_client, db_session):
"""D-05 / D-16: Cloud detail includes capabilities and unsupported_analysis_reason.
The detail endpoint must expose capability information alongside content so
frontend action slots know whether Analyze/Re-analyze is available.
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_cloud_item(
db_session, auth["user"].id, conn.id, analysis_status="pending"
)
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
resp = await async_client.get(url, headers=auth["headers"])
assert resp.status_code == 200, (
f"Expected 200 from cloud detail endpoint, got {resp.status_code}: {resp.text}"
)
body = resp.json()
# Capabilities must be present (D-05)
assert "capabilities" in body, (
"Cloud detail must expose capabilities dict so frontend can render action slots"
)
# unsupported_analysis_reason is optional but must be present as a field (D-16)
# It may be None for supported items
assert "unsupported_analysis_reason" in body, (
"Cloud detail must include unsupported_analysis_reason (D-16)"
)
# ── T-14.1-01: Response excludes forbidden credential/key fields ──────────────
async def test_detail_response_excludes_credentials_and_keys(async_client, db_session):
"""T-14.1-01 / D-18: Cloud detail response must not expose forbidden fields.
The full serialized response body must NOT contain:
- credentials_enc (cloud credential ciphertext)
- object_key (MinIO object path — internal cache detail)
- version_key (internal cache version discriminator)
- password (user credential field)
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_analyzed_item_with_topics(
db_session, auth["user"].id, conn.id
)
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
resp = await async_client.get(url, headers=auth["headers"])
assert resp.status_code == 200, (
f"Expected 200 from cloud detail endpoint, got {resp.status_code}: {resp.text}"
)
# Serialize the full body to string for token presence checks (D-18, T-14.1-01)
body_text = resp.text
# Forbidden tokens — CLAUDE.md allowlist-schema rule
forbidden_tokens = [
"credentials_enc", # cloud credential ciphertext
"object_key", # MinIO internal object path
"version_key", # internal cache version discriminator
]
for token in forbidden_tokens:
assert token not in body_text, (
f"T-14.1-01: Forbidden field '{token}' must not appear in cloud detail response"
)
# Provider URLs must not appear in the body (raw https:// pointing at the provider host)
# A provider URL for google_drive would typically be https://drive.google.com or
# https://www.googleapis.com — the response must use DocuVault-scoped relative URLs only
# (or omit provider URLs entirely).
# We assert no raw http(s) URL pointing to an external provider appears in the body.
# The test checks for the absence of common provider URL substrings that would
# indicate credential-bearing URLs leaking through.
assert "googleapis.com" not in body_text, (
"T-14.1-01: Raw Google API URL must not appear in cloud detail response"
)
assert "graph.microsoft.com" not in body_text, (
"T-14.1-01: Raw Microsoft Graph URL must not appear in cloud detail response"
)
# ── T-14.1-02: IDOR — foreign user gets 404 ──────────────────────────────────
async def test_foreign_user_detail_returns_404(async_client, db_session):
"""T-14.1-02 / D-18: Foreign user GET of another user's cloud item returns 404.
The ownership check (connection_id scoped to current_user.id) must fire before
any content is served. 404 protects against IDOR.
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth_owner = await _create_user_and_token(db_session)
auth_foreign = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth_owner["user"].id)
item = await _create_cloud_item(db_session, auth_owner["user"].id, conn.id)
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
resp = await async_client.get(url, headers=auth_foreign["headers"])
assert resp.status_code == 404, (
f"T-14.1-02: Foreign user must get 404 (IDOR protection) — "
f"got {resp.status_code}: {resp.text}"
)
# ── T-14.1-02: Admin is blocked from cloud detail endpoint ───────────────────
async def test_admin_detail_returns_403(async_client, db_session):
"""T-14.1-02 / D-18: Admin token is blocked from cloud detail by get_regular_user.
The cloud detail endpoint must use get_regular_user, which rejects admin accounts.
Admin accounts must not access user document content.
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth_user = await _create_user_and_token(db_session, role="user")
auth_admin = await _create_user_and_token(db_session, role="admin")
conn = await _create_cloud_connection(db_session, auth_user["user"].id)
item = await _create_cloud_item(db_session, auth_user["user"].id, conn.id)
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
resp = await async_client.get(url, headers=auth_admin["headers"])
assert resp.status_code == 403, (
f"T-14.1-02: Admin must be blocked from cloud detail — "
f"expected 403 from get_regular_user, got {resp.status_code}: {resp.text}"
)
# ── D-07: Stale item returns prior extracted_text and topics ─────────────────
async def test_stale_item_returns_prior_analysis_data(async_client, db_session):
"""D-07: Stale cloud analysis keeps prior extracted text and topics visible.
When analysis_status is 'stale', the detail response must still include
the previously extracted_text and topics — it must not clear them.
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
# Create an item that was indexed but has become stale
item = await _create_analyzed_item_with_topics(
db_session, auth["user"].id, conn.id, topic_names=["legal"]
)
# Update analysis_status to stale (provider changed the file)
from db.models import CloudItem
from sqlalchemy import select as sa_select
result = await db_session.execute(
sa_select(CloudItem).where(CloudItem.id == item.id)
)
item_row = result.scalars().first()
item_row.analysis_status = "stale"
await db_session.commit()
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
resp = await async_client.get(url, headers=auth["headers"])
assert resp.status_code == 200, (
f"Expected 200 for stale item, got {resp.status_code}: {resp.text}"
)
body = resp.json()
# D-07: stale item must still return prior extracted_text and topics
assert "extracted_text" in body, "Stale item must still include extracted_text"
assert body["extracted_text"], "Stale item must retain prior extracted_text"
assert "topics" in body, "Stale item must still include topics"
assert len(body["topics"]) >= 1, "Stale item must retain prior topic links"
# analysis_status must reflect stale state
assert body.get("analysis_status") == "stale", (
f"Stale item must report analysis_status='stale', got {body.get('analysis_status')}"
)
# ── D-05 / D-07: Pending item returns empty analysis fields ──────────────────
async def test_pending_item_returns_empty_analysis_fields(async_client, db_session):
"""D-02 / D-05: Cloud detail exists before analysis and shows empty-state fields.
A cloud file that has not been analyzed yet must return 200 with:
analysis_status = 'pending', extracted_text = None/empty, topics = [].
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_cloud_item(
db_session, auth["user"].id, conn.id, analysis_status="pending"
)
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
resp = await async_client.get(url, headers=auth["headers"])
assert resp.status_code == 200, (
f"Expected 200 for pending item (detail exists before analysis), "
f"got {resp.status_code}: {resp.text}"
)
body = resp.json()
# analysis_status should be pending or none
assert body.get("analysis_status") in ("pending", None, ""), (
f"Expected pending analysis_status, got {body.get('analysis_status')}"
)
# topics must be an empty list (not None — frontend expects a list)
topics = body.get("topics", [])
assert isinstance(topics, list), "topics must be a list even for unanalyzed items"
assert topics == [], f"Unanalyzed item must have empty topics list, got {topics}"
# ── D-18: Detail does not hydrate bytes (no get_object call during detail) ────
async def test_detail_does_not_download_bytes(async_client, db_session):
"""D-18: Cloud detail endpoint must not call adapter.get_object or MinIO.
The detail endpoint returns only metadata from cloud_items — it must not
trigger byte download through the cache or provider adapter.
This test will FAIL until Plan 02 implements the /detail endpoint.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_analyzed_item_with_topics(db_session, auth["user"].id, conn.id)
get_object_tracker = MagicMock(
side_effect=AssertionError("get_object must not be called from cloud detail")
)
url = _DETAIL_PATH_TEMPLATE.format(
connection_id=conn.id,
item_id=item.provider_item_id,
)
with patch("services.cloud_cache.hydrate_and_cache_bytes", side_effect=AssertionError(
"hydrate_and_cache_bytes must not be called from detail endpoint"
)):
resp = await async_client.get(url, headers=auth["headers"])
# If the endpoint exists, it must not have triggered byte hydration
assert resp.status_code == 200, (
f"Expected 200 from cloud detail (metadata-only), got {resp.status_code}: {resp.text}"
)
backend/tests/test_cloud_reanalyze_force.py
+476
View File
@@ -0,0 +1,476 @@
"""
Phase 14.1 Plan 01 — RED contract tests for force re-analyze and single-item retry.
Tests pin the contract for:
1. force=True on the analysis enqueue endpoint (D-11, ANALYZE-06)
2. Single-item retry job creation when no active job exists (D-12, ANALYZE-05)
These tests will FAIL against the current codebase because:
- AnalysisEnqueueRequest has no force field (Plan 02 adds it)
- The single-item retry-with-no-job path may not exist yet
Requirements: ANALYZE-05, ANALYZE-06, ANALYZE-07
Decisions: D-11, D-12, D-18, D-20
"""
from __future__ import annotations
import uuid as _uuid
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
pytestmark = pytest.mark.asyncio
from tests.conftest import _TEST_USER_AGENT
# ── Shared helpers ─────────────────────────────────────────────────────────────
async def _create_user_and_token(session, role: str = "user"):
from db.models import User, Quota
from services.auth import hash_password, create_access_token
user_id = _uuid.uuid4()
user = User(
id=user_id,
handle=f"reana_user_{user_id.hex[:8]}",
email=f"reana_{user_id.hex[:8]}@example.com",
password_hash=hash_password("Testpassword123!"),
role=role,
is_active=True,
password_must_change=False,
)
quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
session.add(user)
session.add(quota)
await session.commit()
await session.refresh(user)
token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
return {
"user": user,
"token": token,
"headers": {
"Authorization": f"Bearer {token}",
"User-Agent": _TEST_USER_AGENT,
},
}
async def _create_cloud_connection(session, user_id, provider="google_drive", status="ACTIVE"):
from db.models import CloudConnection
from storage.cloud_utils import encrypt_credentials
from config import settings
master_key = settings.cloud_creds_key.encode()
creds_enc = encrypt_credentials(
master_key,
str(user_id),
{"access_token": "tok", "refresh_token": "ref"},
)
conn = CloudConnection(
id=_uuid.uuid4(),
user_id=user_id,
provider=provider,
display_name="Force-Analyze Test Connection",
credentials_enc=creds_enc,
status=status,
)
session.add(conn)
await session.commit()
return conn
async def _create_indexed_item(session, user_id, connection_id, etag="etag-indexed-v1"):
"""Create a CloudItem that is already indexed (already_current candidate)."""
from db.models import CloudItem
item = CloudItem(
id=_uuid.uuid4(),
user_id=user_id,
connection_id=connection_id,
provider_item_id=f"pitem-indexed-{_uuid.uuid4().hex[:8]}",
name="already_current.pdf",
kind="file",
content_type="application/pdf",
provider_size=102400,
etag=etag,
analysis_status="indexed",
semantic_index_status="indexed",
extracted_text="Previously extracted document text.",
)
session.add(item)
await session.commit()
return item
async def _create_failed_item(session, user_id, connection_id):
"""Create a CloudItem in 'failed' analysis status (retry candidate)."""
from db.models import CloudItem
item = CloudItem(
id=_uuid.uuid4(),
user_id=user_id,
connection_id=connection_id,
provider_item_id=f"pitem-failed-{_uuid.uuid4().hex[:8]}",
name="failed_analysis.pdf",
kind="file",
content_type="application/pdf",
provider_size=51200,
etag="etag-fail-v1",
analysis_status="failed",
)
session.add(item)
await session.commit()
return item
# ── D-11 / ANALYZE-06: Default enqueue skips already-current items ────────────
async def test_default_enqueue_skips_already_current_item(async_client, db_session):
"""ANALYZE-06 / D-11: Default enqueue marks unchanged indexed items as already_current.
An item with matching version key (same etag/metadata) should yield:
- already_current_count >= 1
- queued_count == 0
This confirms the baseline behavior that force=True overrides.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_indexed_item(db_session, auth["user"].id, conn.id)
adapter = MagicMock()
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
resp = await async_client.post(
f"/api/cloud/analysis/connections/{conn.id}/jobs",
json={
"scope": "file",
"provider_item_ids": [item.provider_item_id],
# No force flag — default idempotent behavior
},
headers=auth["headers"],
)
assert resp.status_code in (200, 202), (
f"Enqueue must succeed — got {resp.status_code}: {resp.text}"
)
body = resp.json()
job_id = body.get("job_id")
assert job_id is not None
# Fetch job status to verify already_current classification
status_resp = await async_client.get(
f"/api/cloud/analysis/jobs/{job_id}",
headers=auth["headers"],
)
assert status_resp.status_code == 200
status_body = status_resp.json()
assert status_body.get("already_current_count", 0) >= 1, (
"Default enqueue must classify unchanged indexed item as already_current"
)
assert status_body.get("queued_count", 0) == 0, (
"Default enqueue must not queue an already-current item"
)
# ── D-11 / ANALYZE-06: force=True enqueues an already-current item ────────────
async def test_force_enqueue_queues_already_current_item(async_client, db_session):
"""ANALYZE-06 / D-11: force=True bypasses already_current and enqueues the item.
The same item that default enqueue skips as already_current must be queued
when the request body includes force=true.
This test WILL FAIL until Plan 02 adds force field to AnalysisEnqueueRequest
and enqueue_analysis_job service.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_indexed_item(db_session, auth["user"].id, conn.id)
adapter = MagicMock()
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
resp = await async_client.post(
f"/api/cloud/analysis/connections/{conn.id}/jobs",
json={
"scope": "file",
"provider_item_ids": [item.provider_item_id],
"force": True, # Plan 02 adds this field to AnalysisEnqueueRequest
},
headers=auth["headers"],
)
# Without Plan 02 this returns 422 (unknown field) or 200 (field ignored)
# either way the assertions below enforce the contract
assert resp.status_code in (200, 202), (
f"Force enqueue must succeed — got {resp.status_code}: {resp.text}. "
"Plan 02 must add force field to AnalysisEnqueueRequest."
)
body = resp.json()
job_id = body.get("job_id")
assert job_id is not None
# Fetch job status
status_resp = await async_client.get(
f"/api/cloud/analysis/jobs/{job_id}",
headers=auth["headers"],
)
assert status_resp.status_code == 200
status_body = status_resp.json()
# With force=True the item must be queued (not already_current)
assert status_body.get("queued_count", 0) >= 1, (
"force=True must enqueue the already-current item for re-analysis (D-11)"
)
assert status_body.get("already_current_count", 0) == 0, (
"force=True must not mark the item as already_current"
)
# ── ANALYZE-06: force flag is present in the AnalysisEnqueueRequest schema ────
def test_force_field_exists_in_enqueue_request_schema():
"""ANALYZE-06 / D-11: AnalysisEnqueueRequest must have a force field.
Plan 02 adds force: bool = False to AnalysisEnqueueRequest.
This test will FAIL until that schema change lands.
"""
from api.cloud.schemas import AnalysisEnqueueRequest
# Check that the schema accepts a 'force' field
fields = AnalysisEnqueueRequest.model_fields
assert "force" in fields, (
"AnalysisEnqueueRequest must have a 'force' field (D-11 / Plan 02). "
"This test fails until Plan 02 adds force: bool = False to the schema."
)
# Default must be False (non-breaking)
default = fields["force"].default
assert default is False, (
f"force must default to False (non-breaking), got default={default!r}"
)
# ── ANALYZE-07: force re-analyze routes through analysis path, no provider mutation
async def test_force_reanalyze_does_not_mutate_provider(async_client, db_session):
"""ANALYZE-07 / D-18: Force re-analyze must not call any provider mutation method.
Even with force=True, the re-analysis pathway must only read bytes (get_object)
and write to the cache/database — it must not call upload, delete, rename,
move, or create_folder on the provider adapter.
This test will FAIL until Plan 02 adds force support to the enqueue endpoint.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_indexed_item(db_session, auth["user"].id, conn.id)
mutation_tracker = MagicMock()
# Adapter that tracks mutation calls — they must stay zero
class NoMutationAdapter:
get_object_calls = 0
mutation_calls = 0
async def get_object(self, *a, **kw):
self.get_object_calls += 1
return b"document content bytes"
async def upload(self, *a, **kw):
self.mutation_calls += 1
raise AssertionError("upload must not be called during force re-analyze")
async def delete(self, *a, **kw):
self.mutation_calls += 1
raise AssertionError("delete must not be called during force re-analyze")
async def rename(self, *a, **kw):
self.mutation_calls += 1
raise AssertionError("rename must not be called during force re-analyze")
async def move(self, *a, **kw):
self.mutation_calls += 1
raise AssertionError("move must not be called during force re-analyze")
async def create_folder(self, *a, **kw):
self.mutation_calls += 1
raise AssertionError("create_folder must not be called during force re-analyze")
adapter = NoMutationAdapter()
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
resp = await async_client.post(
f"/api/cloud/analysis/connections/{conn.id}/jobs",
json={
"scope": "file",
"provider_item_ids": [item.provider_item_id],
"force": True,
},
headers=auth["headers"],
)
assert resp.status_code in (200, 202), (
f"Force enqueue must succeed — got {resp.status_code}: {resp.text}"
)
# The enqueue path must not have called any provider mutation
assert adapter.mutation_calls == 0, (
f"ANALYZE-07: force enqueue called {adapter.mutation_calls} provider mutation(s)"
)
# ── D-12 / ANALYZE-05: Retry with no active job creates a single-item retry job
async def test_retry_failed_item_with_no_active_job_creates_single_item_job(
async_client, db_session
):
"""D-12 / ANALYZE-05: Retry a failed item when no active job exists.
If a cloud item has analysis_status='failed' and no active/queued job covers it,
retrying via the detail or analysis retry endpoint must create a single-item job
(or return a typed result that yields exactly one queued item).
This test will FAIL until Plan 02 adds the owner-scoped single-item retry endpoint
or extends the existing retry to create a new job when none exists.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_failed_item(db_session, auth["user"].id, conn.id)
# No active job exists for this item — retry must create one
# Target path: POST /api/cloud/analysis/connections/{conn_id}/items/{item_id}/retry
# (Plan 02 adds this route or extends the existing retry semantics)
retry_url = f"/api/cloud/analysis/connections/{conn.id}/items/{item.id}/retry"
adapter = MagicMock()
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
resp = await async_client.post(retry_url, headers=auth["headers"])
assert resp.status_code in (200, 202), (
f"Single-item retry with no active job must succeed — "
f"got {resp.status_code}: {resp.text}. "
"Plan 02 must implement this route or extend retry semantics."
)
body = resp.json()
# The response must indicate exactly one item was queued
# Either via a new job (job_id in response) or a typed retry result
has_job_id = "job_id" in body
has_queued = body.get("queued_count", 0) >= 1
assert has_job_id or has_queued, (
f"Single-item retry must return a job_id or queued_count >= 1, "
f"got: {body}"
)
if has_job_id:
# If a job was created, verify it has exactly one queued item
status_resp = await async_client.get(
f"/api/cloud/analysis/jobs/{body['job_id']}",
headers=auth["headers"],
)
assert status_resp.status_code == 200
status_body = status_resp.json()
assert status_body.get("queued_count", 0) >= 1, (
"Single-item retry job must have at least one queued item"
)
# ── T-14.1-02: Force enqueue is owner-scoped ──────────────────────────────────
async def test_force_enqueue_foreign_user_blocked(async_client, db_session):
"""T-14.1-02: Force re-analyze must be owner-scoped — foreign user gets 403/404."""
auth_owner = await _create_user_and_token(db_session)
auth_foreign = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth_owner["user"].id)
item = await _create_indexed_item(db_session, auth_owner["user"].id, conn.id)
adapter = MagicMock()
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
resp = await async_client.post(
f"/api/cloud/analysis/connections/{conn.id}/jobs",
json={
"scope": "file",
"provider_item_ids": [item.provider_item_id],
"force": True,
},
headers=auth_foreign["headers"],
)
assert resp.status_code in (403, 404), (
f"T-14.1-02: Force enqueue by foreign user must be blocked — "
f"got {resp.status_code}"
)
async def test_force_enqueue_admin_blocked(async_client, db_session):
"""T-14.1-02: Admin accounts cannot force enqueue analysis (get_regular_user blocks)."""
auth_user = await _create_user_and_token(db_session, role="user")
auth_admin = await _create_user_and_token(db_session, role="admin")
conn = await _create_cloud_connection(db_session, auth_user["user"].id)
item = await _create_indexed_item(db_session, auth_user["user"].id, conn.id)
adapter = MagicMock()
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
resp = await async_client.post(
f"/api/cloud/analysis/connections/{conn.id}/jobs",
json={
"scope": "file",
"provider_item_ids": [item.provider_item_id],
"force": True,
},
headers=auth_admin["headers"],
)
assert resp.status_code in (403, 404), (
f"T-14.1-02: Admin force enqueue must be blocked by get_regular_user — "
f"got {resp.status_code}"
)
# ── ANALYZE-06: force=false behaves identically to omitted force ───────────────
async def test_force_false_is_equivalent_to_default_enqueue(async_client, db_session):
"""ANALYZE-06: Explicit force=false is identical to default (non-forced) enqueue.
force=False must preserve normal idempotent already-current behavior.
"""
auth = await _create_user_and_token(db_session)
conn = await _create_cloud_connection(db_session, auth["user"].id)
item = await _create_indexed_item(db_session, auth["user"].id, conn.id)
adapter = MagicMock()
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
resp = await async_client.post(
f"/api/cloud/analysis/connections/{conn.id}/jobs",
json={
"scope": "file",
"provider_item_ids": [item.provider_item_id],
"force": False, # Explicit false — same as default
},
headers=auth["headers"],
)
# Even if force field doesn't exist yet (422), the already-current behavior is tested
# by test_default_enqueue_skips_already_current_item above.
# This test verifies force=False is accepted (not rejected as unknown field) after Plan 02.
assert resp.status_code in (200, 202, 422), (
f"force=False enqueue returned unexpected status: {resp.status_code}: {resp.text}"
)
if resp.status_code in (200, 202):
body = resp.json()
job_id = body.get("job_id")
if job_id:
status_resp = await async_client.get(
f"/api/cloud/analysis/jobs/{job_id}",
headers=auth["headers"],
)
if status_resp.status_code == 200:
status_body = status_resp.json()
# With force=False, already_current item should NOT be queued
assert status_body.get("queued_count", 0) == 0, (
"force=False must not queue an already-current item"
)