curo1305
f235d894fb
- Implement tasks/cloud_analysis_tasks.py: sync Celery bridge to async processing - Broker payload contains IDs only — credentials decrypted in worker (T-14-10) - _TransientError and _ClassificationRetry sentinels for bounded retry (max=3) - Auth errors are terminal (not retried); MaxRetriesExceeded writes failure status - celery_app.py: add cloud_analysis_tasks route and import registration - Add 6 contract tests: task registration, broker payload shape, queue routing, bounded retries, sentinel pattern, connection-not-found graceful drop
1242 lines
47 KiB
Python
1242 lines
47 KiB
Python
"""
|
|
Phase 14 Plan 01 — RED contract tests for cloud analysis scope, estimates, and
|
|
idempotency (no bytes downloaded during estimate or already-current checks).
|
|
|
|
Requirements: ANALYZE-01, ANALYZE-02, ANALYZE-03, ANALYZE-06, ANALYZE-07
|
|
|
|
Threat coverage:
|
|
T-14-01 — IDOR: foreign user cannot estimate/enqueue another user's connection
|
|
T-14-03 — Hidden provider mutation: fake adapter mutation counters stay zero
|
|
T-14-04 — Hidden byte download: estimate and already-current checks never call
|
|
provider byte fetch
|
|
|
|
All tests import non-existent Phase 14 modules and must FAIL due to
|
|
ImportError or missing endpoints. They define the implementation contract.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
from tests.conftest import _TEST_USER_AGENT
|
|
|
|
|
|
# ─── Minimal fake adapter ────────────────────────────────────────────────────
|
|
|
|
class FakeCloudAdapter:
|
|
"""Counts calls to mutation and byte-download methods.
|
|
|
|
Zero counts on all methods after analysis operations confirms
|
|
T-14-03 (no mutation) and T-14-04 (no byte download).
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.upload_calls = 0
|
|
self.delete_calls = 0
|
|
self.rename_calls = 0
|
|
self.move_calls = 0
|
|
self.create_folder_calls = 0
|
|
self.get_object_calls = 0 # byte download
|
|
self.put_object_calls = 0 # byte upload
|
|
|
|
# Mutations — must stay zero during analysis
|
|
async def upload(self, *a, **kw):
|
|
self.upload_calls += 1
|
|
raise AssertionError("upload must not be called during analysis")
|
|
|
|
async def delete(self, *a, **kw):
|
|
self.delete_calls += 1
|
|
raise AssertionError("delete must not be called during analysis")
|
|
|
|
async def rename(self, *a, **kw):
|
|
self.rename_calls += 1
|
|
raise AssertionError("rename must not be called during analysis")
|
|
|
|
async def move(self, *a, **kw):
|
|
self.move_calls += 1
|
|
raise AssertionError("move must not be called during analysis")
|
|
|
|
async def create_folder(self, *a, **kw):
|
|
self.create_folder_calls += 1
|
|
raise AssertionError("create_folder must not be called during analysis")
|
|
|
|
# Byte downloads — must stay zero during estimate and already-current checks
|
|
async def get_object(self, *a, **kw):
|
|
self.get_object_calls += 1
|
|
raise AssertionError("get_object must not be called during estimate or already-current check")
|
|
|
|
async def put_object(self, *a, **kw):
|
|
self.put_object_calls += 1
|
|
raise AssertionError("put_object must not be called during analysis path")
|
|
|
|
@property
|
|
def mutation_call_count(self):
|
|
return (
|
|
self.upload_calls + self.delete_calls + self.rename_calls +
|
|
self.move_calls + self.create_folder_calls
|
|
)
|
|
|
|
@property
|
|
def byte_call_count(self):
|
|
return self.get_object_calls + self.put_object_calls
|
|
|
|
|
|
# ─── Shared test 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"ana_user_{user_id.hex[:8]}",
|
|
email=f"ana_{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="Test Connection",
|
|
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="doc.pdf",
|
|
kind="file", etag="etag-v1", analysis_status="pending"):
|
|
from db.models import CloudItem
|
|
|
|
item = CloudItem(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
connection_id=connection_id,
|
|
provider_item_id=f"pitem-{uuid.uuid4().hex[:8]}",
|
|
name=name,
|
|
kind=kind,
|
|
content_type="application/pdf",
|
|
provider_size=102400,
|
|
etag=etag,
|
|
analysis_status=analysis_status,
|
|
)
|
|
session.add(item)
|
|
await session.commit()
|
|
return item
|
|
|
|
|
|
# ─── ANALYZE-01: Single file analysis estimate ───────────────────────────────
|
|
|
|
async def test_estimate_single_file_returns_supported_count_and_bytes(async_client, db_session):
|
|
"""ANALYZE-01: Single-file estimate returns count=1 and provider_bytes without downloading.
|
|
|
|
T-14-04: no bytes fetched during estimate.
|
|
"""
|
|
# Requires Phase 14 endpoint: POST /api/cloud/analysis/connections/{id}/estimate
|
|
from api.cloud.analysis import router as analysis_router # Phase 14 — not yet implemented
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/estimate",
|
|
json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["supported_count"] == 1
|
|
assert body["unsupported_count"] == 0
|
|
assert "total_provider_bytes" in body
|
|
assert body["total_provider_bytes"] == item.provider_size
|
|
|
|
# T-14-04: no byte downloads during estimate
|
|
assert adapter.byte_call_count == 0, "Estimate must not download provider bytes"
|
|
# T-14-03: no mutations during estimate
|
|
assert adapter.mutation_call_count == 0, "Estimate must not call any provider mutation"
|
|
|
|
|
|
# ─── ANALYZE-02: Selection estimate ─────────────────────────────────────────
|
|
|
|
async def test_estimate_selection_aggregates_multiple_files(async_client, db_session):
|
|
"""ANALYZE-02: Multi-file selection estimate aggregates counts and bytes.
|
|
|
|
T-14-04: bytes not fetched during selection estimate.
|
|
"""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
item1 = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="a.pdf")
|
|
item2 = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="b.docx")
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/estimate",
|
|
json={
|
|
"scope": "selection",
|
|
"provider_item_ids": [item1.provider_item_id, item2.provider_item_id],
|
|
},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["supported_count"] == 2
|
|
# total bytes = sum of provider sizes
|
|
assert body["total_provider_bytes"] == (item1.provider_size or 0) + (item2.provider_size or 0)
|
|
|
|
assert adapter.byte_call_count == 0
|
|
assert adapter.mutation_call_count == 0
|
|
|
|
|
|
async def test_estimate_includes_unsupported_item_count(async_client, db_session):
|
|
"""ANALYZE-02: Estimate reports unsupported items without blocking supportable ones.
|
|
|
|
Unsupported items are reported but do not cause estimate to fail (D-23).
|
|
"""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
pdf_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="report.pdf")
|
|
# An item with an unsupported content type (e.g. .exe) should count as unsupported
|
|
exe_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="setup.exe")
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/estimate",
|
|
json={
|
|
"scope": "selection",
|
|
"provider_item_ids": [pdf_item.provider_item_id, exe_item.provider_item_id],
|
|
},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
# At least one supported (pdf), at least one unsupported (exe)
|
|
assert body["supported_count"] >= 1
|
|
assert body["unsupported_count"] >= 1
|
|
assert adapter.byte_call_count == 0
|
|
|
|
|
|
# ─── ANALYZE-03: Whole-connection estimate ───────────────────────────────────
|
|
|
|
async def test_whole_connection_estimate_always_required(async_client, db_session):
|
|
"""ANALYZE-03: Whole-connection scope always produces an estimate response before enqueue.
|
|
|
|
The estimate endpoint must be called before enqueue for connection-scope (D-02).
|
|
T-14-04: bytes not fetched.
|
|
"""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="file1.pdf")
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/estimate",
|
|
json={"scope": "connection", "recursive": True},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
# Whole-connection always includes a recursive flag in the response
|
|
assert "recursive" in body or body.get("scope_kind") == "connection"
|
|
assert "supported_count" in body
|
|
assert adapter.byte_call_count == 0
|
|
|
|
|
|
async def test_folder_estimate_asks_about_recursion(async_client, db_session):
|
|
"""ANALYZE-03/D-04: Folder-scope estimate accepts recursive flag and reports it back."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
folder_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id,
|
|
name="Reports", kind="folder")
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/estimate",
|
|
json={
|
|
"scope": "folder",
|
|
"provider_item_ids": [folder_item.provider_item_id],
|
|
"recursive": True,
|
|
},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "recursive" in body
|
|
assert adapter.byte_call_count == 0
|
|
|
|
|
|
# ─── ANALYZE-06: Duplicate / already-current skip ────────────────────────────
|
|
|
|
async def test_already_current_item_skipped_without_byte_fetch(async_client, db_session):
|
|
"""ANALYZE-06: Item with unchanged etag is skipped without downloading provider bytes.
|
|
|
|
T-14-04: get_object must not be called for already-current items.
|
|
"""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
# Item already indexed with same etag
|
|
item = await _create_cloud_item(
|
|
db_session, ctx["user"].id, conn.id,
|
|
etag="etag-stable-v1", analysis_status="indexed"
|
|
)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
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],
|
|
},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
# Enqueue succeeds (202 Accepted or 200 with job status)
|
|
assert resp.status_code in (200, 202)
|
|
body = resp.json()
|
|
|
|
# Item reported as already_current — not re-downloaded
|
|
# The job should report 0 queued items (or the item appears as already_current)
|
|
job_id = body.get("job_id")
|
|
assert job_id is not None
|
|
|
|
# Fetch job status to verify already_current
|
|
status_resp = await async_client.get(
|
|
f"/api/cloud/analysis/jobs/{job_id}",
|
|
headers=ctx["headers"],
|
|
)
|
|
assert status_resp.status_code == 200
|
|
status_body = status_resp.json()
|
|
|
|
# Item should not be in "queued" state — it's already current
|
|
assert status_body.get("already_current_count", 0) >= 1
|
|
assert status_body.get("queued_count", 0) == 0
|
|
|
|
# Critical: no bytes were fetched
|
|
assert adapter.byte_call_count == 0, "already-current check must not download provider bytes"
|
|
# Critical: no mutations occurred
|
|
assert adapter.mutation_call_count == 0
|
|
|
|
|
|
async def test_changed_etag_triggers_reanalysis(async_client, db_session):
|
|
"""ANALYZE-06: Item with changed etag is queued for reanalysis (not skipped)."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
# Item previously indexed but etag has changed
|
|
item = await _create_cloud_item(
|
|
db_session, ctx["user"].id, conn.id,
|
|
etag="etag-OLD", analysis_status="indexed"
|
|
)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
# Provider now reports a different etag
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
with patch("services.cloud_analysis.resolve_provider_metadata",
|
|
new_callable=AsyncMock,
|
|
return_value={"etag": "etag-NEW", "size": 102400}):
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code in (200, 202)
|
|
body = resp.json()
|
|
job_id = body.get("job_id")
|
|
assert job_id is not None
|
|
|
|
status_resp = await async_client.get(
|
|
f"/api/cloud/analysis/jobs/{job_id}",
|
|
headers=ctx["headers"],
|
|
)
|
|
assert status_resp.status_code == 200
|
|
status_body = status_resp.json()
|
|
# Item is queued for re-analysis (not marked already_current)
|
|
assert status_body.get("queued_count", 0) >= 1
|
|
|
|
|
|
# ─── ANALYZE-07: Provider file not mutated ───────────────────────────────────
|
|
|
|
async def test_analysis_never_mutates_provider_file(async_client, db_session):
|
|
"""ANALYZE-07: Analysis job completion never calls any provider mutation method.
|
|
|
|
T-14-03: mutation counters must remain zero for the full analysis lifecycle.
|
|
"""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
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]},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code in (200, 202)
|
|
# No mutations at any point in analysis
|
|
assert adapter.mutation_call_count == 0, (
|
|
f"Analysis called provider mutations: upload={adapter.upload_calls}, "
|
|
f"delete={adapter.delete_calls}, rename={adapter.rename_calls}, "
|
|
f"move={adapter.move_calls}, create_folder={adapter.create_folder_calls}"
|
|
)
|
|
|
|
|
|
# ─── T-14-01: IDOR — foreign user cannot estimate/enqueue ───────────────────
|
|
|
|
async def test_foreign_user_cannot_estimate_another_users_connection(async_client, db_session):
|
|
"""T-14-01: Estimate request against another user's connection returns 403/404."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
owner_ctx = await _create_user_and_token(db_session)
|
|
foreign_ctx = await _create_user_and_token(db_session)
|
|
owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
|
|
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{owner_conn.id}/estimate",
|
|
json={"scope": "connection", "recursive": True},
|
|
headers=foreign_ctx["headers"], # Foreign user's token
|
|
)
|
|
|
|
assert resp.status_code in (403, 404), (
|
|
f"Foreign user estimate must be rejected — got {resp.status_code}"
|
|
)
|
|
|
|
|
|
async def test_foreign_user_cannot_enqueue_another_users_connection(async_client, db_session):
|
|
"""T-14-01: Enqueue request against another user's connection is rejected."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
owner_ctx = await _create_user_and_token(db_session)
|
|
foreign_ctx = await _create_user_and_token(db_session)
|
|
owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
|
|
owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)
|
|
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
|
|
headers=foreign_ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code in (403, 404), (
|
|
f"Foreign user enqueue must be rejected — got {resp.status_code}"
|
|
)
|
|
|
|
|
|
async def test_admin_cannot_enqueue_another_users_analysis_job(async_client, db_session):
|
|
"""T-14-01: Admin role cannot access cloud analysis routes (same as browse restriction)."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
user_ctx = await _create_user_and_token(db_session, role="user")
|
|
admin_ctx = await _create_user_and_token(db_session, role="admin")
|
|
user_conn = await _create_cloud_connection(db_session, user_ctx["user"].id)
|
|
user_item = await _create_cloud_item(db_session, user_ctx["user"].id, user_conn.id)
|
|
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{user_conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [user_item.provider_item_id]},
|
|
headers=admin_ctx["headers"],
|
|
)
|
|
|
|
assert resp.status_code in (403, 404), (
|
|
f"Admin must not access cloud analysis — got {resp.status_code}"
|
|
)
|
|
|
|
|
|
# ─── Job status and controls ─────────────────────────────────────────────────
|
|
|
|
async def test_job_status_response_excludes_credentials_and_object_keys(async_client, db_session):
|
|
"""T-14-02: Job status response never includes credentials or raw object keys."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
enqueue_resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
assert enqueue_resp.status_code in (200, 202)
|
|
job_id = enqueue_resp.json().get("job_id")
|
|
assert job_id is not None
|
|
|
|
status_resp = await async_client.get(
|
|
f"/api/cloud/analysis/jobs/{job_id}",
|
|
headers=ctx["headers"],
|
|
)
|
|
assert status_resp.status_code == 200
|
|
body_text = status_resp.text
|
|
|
|
# T-14-02: No credential fields or raw object keys in response
|
|
forbidden_fields = [
|
|
"credentials_enc", "access_token", "refresh_token",
|
|
"client_secret", "object_key",
|
|
]
|
|
for field in forbidden_fields:
|
|
assert field not in body_text, (
|
|
f"Sensitive field '{field}' must not appear in job status response"
|
|
)
|
|
|
|
|
|
async def test_foreign_user_cannot_read_job_status(async_client, db_session):
|
|
"""T-14-01: Foreign user cannot read job status for another user's job."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
owner_ctx = await _create_user_and_token(db_session)
|
|
foreign_ctx = await _create_user_and_token(db_session)
|
|
owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
|
|
owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
enqueue_resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
|
|
headers=owner_ctx["headers"],
|
|
)
|
|
|
|
job_id = enqueue_resp.json().get("job_id")
|
|
|
|
# Foreign user tries to read the job
|
|
resp = await async_client.get(
|
|
f"/api/cloud/analysis/jobs/{job_id}",
|
|
headers=foreign_ctx["headers"],
|
|
)
|
|
assert resp.status_code in (403, 404)
|
|
|
|
|
|
async def test_foreign_user_cannot_cancel_another_users_job(async_client, db_session):
|
|
"""T-14-01: Foreign user cannot cancel a job owned by another user."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
owner_ctx = await _create_user_and_token(db_session)
|
|
foreign_ctx = await _create_user_and_token(db_session)
|
|
owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
|
|
owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
enqueue_resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
|
|
headers=owner_ctx["headers"],
|
|
)
|
|
|
|
job_id = enqueue_resp.json().get("job_id")
|
|
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/jobs/{job_id}/cancel",
|
|
headers=foreign_ctx["headers"],
|
|
)
|
|
assert resp.status_code in (403, 404)
|
|
|
|
|
|
# ─── Cancel and retry state transitions ──────────────────────────────────────
|
|
|
|
async def test_cancel_job_transitions_to_cancelled_status(async_client, db_session):
|
|
"""ANALYZE-05: Cancel job transitions the job status to 'cancelled'."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
enqueue_resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
job_id = enqueue_resp.json().get("job_id")
|
|
|
|
cancel_resp = await async_client.post(
|
|
f"/api/cloud/analysis/jobs/{job_id}/cancel",
|
|
headers=ctx["headers"],
|
|
)
|
|
assert cancel_resp.status_code in (200, 202, 204)
|
|
|
|
status_resp = await async_client.get(
|
|
f"/api/cloud/analysis/jobs/{job_id}",
|
|
headers=ctx["headers"],
|
|
)
|
|
assert status_resp.json().get("status") in ("cancelled", "cancelling")
|
|
|
|
|
|
async def test_retry_failed_item_requeues_it(async_client, db_session):
|
|
"""ANALYZE-05: Retry on a failed item transitions it back to queued status."""
|
|
from api.cloud.analysis import router as analysis_router # Phase 14
|
|
|
|
ctx = await _create_user_and_token(db_session)
|
|
conn = await _create_cloud_connection(db_session, ctx["user"].id)
|
|
item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)
|
|
|
|
adapter = FakeCloudAdapter()
|
|
with patch("services.cloud_analysis.get_adapter", return_value=adapter):
|
|
enqueue_resp = await async_client.post(
|
|
f"/api/cloud/analysis/connections/{conn.id}/jobs",
|
|
json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
|
|
headers=ctx["headers"],
|
|
)
|
|
|
|
job_id = enqueue_resp.json().get("job_id")
|
|
|
|
# Simulate item in failed state then retry
|
|
resp = await async_client.post(
|
|
f"/api/cloud/analysis/jobs/{job_id}/items/{item.id}/retry",
|
|
headers=ctx["headers"],
|
|
)
|
|
# Must return a valid status that indicates retry was accepted
|
|
assert resp.status_code in (200, 202, 204)
|
|
|
|
|
|
# ─── Plan 05: processing service contract ────────────────────────────────────
|
|
|
|
async def test_processing_status_transitions_include_downloading_extracting_classifying(db_session):
|
|
"""ANALYZE-04: Processing service transitions through correct intermediate states.
|
|
|
|
The processing service must update item status through the full pipeline:
|
|
queued → downloading → extracting → classifying → indexed.
|
|
"""
|
|
from services.cloud_analysis_processing import process_job_item, ProcessingResult
|
|
|
|
# Verify the ProcessingResult dataclass exists and has the expected fields
|
|
import inspect
|
|
fields = {f for f in ProcessingResult.__dataclass_fields__}
|
|
assert "status" in fields
|
|
assert "cache_entry_id" in fields
|
|
assert "error_code" in fields
|
|
assert "error_message" in fields
|
|
|
|
|
|
async def test_processing_unchanged_version_skips_before_provider_byte_fetch(db_session):
|
|
"""ANALYZE-06 / T-14-13: Stale version detection fires before provider byte fetch.
|
|
|
|
If the version key at processing time differs from the enqueued version key,
|
|
the item must be marked stale WITHOUT downloading provider bytes.
|
|
"""
|
|
from services.cloud_analysis_processing import process_job_item, ProcessingResult
|
|
from services.cloud_analysis import enqueue_analysis_job, AnalysisJobNotFound
|
|
from db.models import CloudAnalysisJobItem
|
|
|
|
# Setup
|
|
from db.models import User, Quota, CloudConnection, CloudItem
|
|
from services.auth import hash_password
|
|
from storage.cloud_utils import encrypt_credentials
|
|
from config import settings
|
|
|
|
user_id = uuid.uuid4()
|
|
user = User(
|
|
id=user_id,
|
|
handle=f"proc_user_{user_id.hex[:8]}",
|
|
email=f"proc_{user_id.hex[:8]}@example.com",
|
|
password_hash=hash_password("Testpassword123!"),
|
|
role="user",
|
|
is_active=True,
|
|
password_must_change=False,
|
|
)
|
|
quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
|
|
db_session.add(user)
|
|
db_session.add(quota)
|
|
await db_session.commit()
|
|
|
|
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="google_drive",
|
|
display_name="Proc Test",
|
|
credentials_enc=creds_enc,
|
|
status="ACTIVE",
|
|
)
|
|
db_session.add(conn)
|
|
await db_session.commit()
|
|
|
|
item = CloudItem(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
provider_item_id=f"pitem-{uuid.uuid4().hex[:8]}",
|
|
name="doc.pdf",
|
|
kind="file",
|
|
content_type="application/pdf",
|
|
provider_size=102400,
|
|
etag="etag-v1",
|
|
analysis_status="pending",
|
|
)
|
|
db_session.add(item)
|
|
await db_session.commit()
|
|
|
|
# Enqueue job
|
|
result = await enqueue_analysis_job(
|
|
db_session,
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
scope="file",
|
|
provider_item_ids=[item.provider_item_id],
|
|
)
|
|
job_id = result.job_id
|
|
|
|
# Fetch the job item
|
|
from sqlalchemy import select as sa_select
|
|
from db.models import CloudAnalysisJob
|
|
item_row = (await db_session.execute(
|
|
sa_select(CloudAnalysisJobItem).where(
|
|
CloudAnalysisJobItem.job_id == job_id,
|
|
CloudAnalysisJobItem.cloud_item_id == item.id,
|
|
)
|
|
)).scalars().first()
|
|
assert item_row is not None
|
|
|
|
# Simulate item metadata changing: update etag on the CloudItem row
|
|
item.etag = "etag-v2-CHANGED"
|
|
item.updated_at = __import__("datetime").datetime.now(__import__("datetime").timezone.utc)
|
|
await db_session.flush()
|
|
await db_session.commit()
|
|
|
|
# FakeAdapter — get_object must NOT be called (version changed before download)
|
|
get_object_calls = []
|
|
|
|
class StrictFakeAdapter:
|
|
async def get_object(self, *a, **kw):
|
|
get_object_calls.append(1)
|
|
raise AssertionError("get_object must not be called for stale item")
|
|
|
|
class FakeMinIO:
|
|
async def get_object(self, key):
|
|
raise AssertionError("MinIO get_object must not be called for stale item")
|
|
async def put_object(self, *a, **kw):
|
|
pass
|
|
|
|
proc_result = await process_job_item(
|
|
db_session,
|
|
job_id=job_id,
|
|
item_id=item_row.id,
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
cloud_item_id=item.id,
|
|
minio_client=FakeMinIO(),
|
|
provider_adapter=StrictFakeAdapter(),
|
|
)
|
|
|
|
# Must be marked stale — no bytes fetched
|
|
assert proc_result.status == "stale"
|
|
assert len(get_object_calls) == 0, "get_object must not be called for stale item"
|
|
|
|
|
|
async def test_processing_cache_pin_released_on_cancellation(db_session):
|
|
"""T-14-12: Cache pin is released when processing is cancelled cooperatively."""
|
|
from services.cloud_analysis_processing import process_job_item
|
|
from services.cloud_analysis import enqueue_analysis_job
|
|
from services.cloud_cache import create_cache_entry, list_cache_entries
|
|
from db.models import CloudAnalysisJob, CloudAnalysisJobItem, CloudConnection, CloudItem
|
|
from services.auth import hash_password
|
|
from storage.cloud_utils import encrypt_credentials
|
|
from config import settings
|
|
|
|
user_id = uuid.uuid4()
|
|
from db.models import User, Quota
|
|
user = User(
|
|
id=user_id,
|
|
handle=f"pin_user_{user_id.hex[:8]}",
|
|
email=f"pin_{user_id.hex[:8]}@example.com",
|
|
password_hash=hash_password("Testpassword123!"),
|
|
role="user",
|
|
is_active=True,
|
|
password_must_change=False,
|
|
)
|
|
quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
|
|
db_session.add(user)
|
|
db_session.add(quota)
|
|
await db_session.commit()
|
|
|
|
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="google_drive",
|
|
display_name="Pin Test Conn",
|
|
credentials_enc=creds_enc,
|
|
status="ACTIVE",
|
|
)
|
|
db_session.add(conn)
|
|
await db_session.commit()
|
|
|
|
item = CloudItem(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
provider_item_id=f"pitem-pin-{uuid.uuid4().hex[:8]}",
|
|
name="doc.txt",
|
|
kind="file",
|
|
content_type="text/plain",
|
|
provider_size=1024,
|
|
etag="etag-pin-v1",
|
|
analysis_status="pending",
|
|
)
|
|
db_session.add(item)
|
|
await db_session.commit()
|
|
|
|
result = await enqueue_analysis_job(
|
|
db_session,
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
scope="file",
|
|
provider_item_ids=[item.provider_item_id],
|
|
)
|
|
job_id = result.job_id
|
|
|
|
from sqlalchemy import select as sa_select
|
|
job = (await db_session.execute(
|
|
sa_select(CloudAnalysisJob).where(CloudAnalysisJob.id == job_id)
|
|
)).scalars().first()
|
|
|
|
# Pre-cancel the job so cancellation triggers immediately
|
|
job.status = "cancelled"
|
|
await db_session.commit()
|
|
|
|
item_row = (await db_session.execute(
|
|
sa_select(CloudAnalysisJobItem).where(
|
|
CloudAnalysisJobItem.job_id == job_id,
|
|
CloudAnalysisJobItem.cloud_item_id == item.id,
|
|
)
|
|
)).scalars().first()
|
|
|
|
class FakeMinIO:
|
|
async def get_object(self, key):
|
|
return b"test content"
|
|
async def put_object(self, *a, **kw):
|
|
pass
|
|
|
|
class FakeAdapter:
|
|
async def get_object(self, *a, **kw):
|
|
return b"test content"
|
|
|
|
proc_result = await process_job_item(
|
|
db_session,
|
|
job_id=job_id,
|
|
item_id=item_row.id,
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
cloud_item_id=item.id,
|
|
minio_client=FakeMinIO(),
|
|
provider_adapter=FakeAdapter(),
|
|
)
|
|
|
|
# Cancelled — no bytes should have been downloaded (cancellation before download)
|
|
assert proc_result.status == "cancelled"
|
|
|
|
# Verify no pinned cache entries exist (pin was not created or was released)
|
|
from db.models import CloudByteCacheEntry
|
|
entries = (await db_session.execute(
|
|
sa_select(CloudByteCacheEntry).where(
|
|
CloudByteCacheEntry.user_id == user_id,
|
|
CloudByteCacheEntry.pin_count > 0,
|
|
)
|
|
)).scalars().all()
|
|
assert len(entries) == 0, "No pinned cache entries must remain after cancellation (T-14-12)"
|
|
|
|
|
|
async def test_processing_never_calls_provider_mutation_methods(db_session):
|
|
"""T-14-03: Processing service must not call any provider mutation methods.
|
|
|
|
The FakeAdapter tracks mutation call counts; all must be zero after processing.
|
|
"""
|
|
from services.cloud_analysis_processing import process_job_item
|
|
from services.cloud_analysis import enqueue_analysis_job
|
|
from db.models import User, Quota, CloudConnection, CloudItem, CloudAnalysisJobItem
|
|
from services.auth import hash_password
|
|
from storage.cloud_utils import encrypt_credentials
|
|
from config import settings
|
|
|
|
user_id = uuid.uuid4()
|
|
user = User(
|
|
id=user_id,
|
|
handle=f"mut_user_{user_id.hex[:8]}",
|
|
email=f"mut_{user_id.hex[:8]}@example.com",
|
|
password_hash=hash_password("Testpassword123!"),
|
|
role="user",
|
|
is_active=True,
|
|
password_must_change=False,
|
|
)
|
|
quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
|
|
db_session.add(user)
|
|
db_session.add(quota)
|
|
await db_session.commit()
|
|
|
|
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="google_drive",
|
|
display_name="Mut Test",
|
|
credentials_enc=creds_enc,
|
|
status="ACTIVE",
|
|
)
|
|
db_session.add(conn)
|
|
await db_session.commit()
|
|
|
|
item = CloudItem(
|
|
id=uuid.uuid4(),
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
provider_item_id=f"pitem-mut-{uuid.uuid4().hex[:8]}",
|
|
name="report.txt",
|
|
kind="file",
|
|
content_type="text/plain",
|
|
provider_size=512,
|
|
etag="etag-mut-v1",
|
|
analysis_status="pending",
|
|
)
|
|
db_session.add(item)
|
|
await db_session.commit()
|
|
|
|
result = await enqueue_analysis_job(
|
|
db_session,
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
scope="file",
|
|
provider_item_ids=[item.provider_item_id],
|
|
)
|
|
job_id = result.job_id
|
|
|
|
from sqlalchemy import select as sa_select
|
|
item_row = (await db_session.execute(
|
|
sa_select(CloudAnalysisJobItem).where(
|
|
CloudAnalysisJobItem.job_id == job_id,
|
|
CloudAnalysisJobItem.cloud_item_id == item.id,
|
|
)
|
|
)).scalars().first()
|
|
|
|
adapter = FakeCloudAdapter()
|
|
# Override get_object to return bytes (not raise) — processing should succeed
|
|
file_content = b"Test document text for classification."
|
|
|
|
class ReadOnlyAdapter:
|
|
def __init__(self):
|
|
self.upload_calls = 0
|
|
self.delete_calls = 0
|
|
self.rename_calls = 0
|
|
self.move_calls = 0
|
|
self.create_folder_calls = 0
|
|
self.get_object_calls = 0
|
|
|
|
async def get_object(self, *a, **kw):
|
|
self.get_object_calls += 1
|
|
return file_content
|
|
|
|
async def upload(self, *a, **kw):
|
|
self.upload_calls += 1
|
|
raise AssertionError("upload must not be called during processing")
|
|
|
|
async def delete(self, *a, **kw):
|
|
self.delete_calls += 1
|
|
raise AssertionError("delete must not be called during processing")
|
|
|
|
async def rename(self, *a, **kw):
|
|
self.rename_calls += 1
|
|
raise AssertionError("rename must not be called during processing")
|
|
|
|
async def move(self, *a, **kw):
|
|
self.move_calls += 1
|
|
raise AssertionError("move must not be called during processing")
|
|
|
|
async def create_folder(self, *a, **kw):
|
|
self.create_folder_calls += 1
|
|
raise AssertionError("create_folder must not be called during processing")
|
|
|
|
@property
|
|
def mutation_call_count(self):
|
|
return (
|
|
self.upload_calls + self.delete_calls + self.rename_calls +
|
|
self.move_calls + self.create_folder_calls
|
|
)
|
|
|
|
read_only = ReadOnlyAdapter()
|
|
|
|
class FakeMinIO:
|
|
async def get_object(self, key):
|
|
raise ValueError("no cache hit")
|
|
async def put_object(self, key, data, content_type=None):
|
|
pass # No-op — quota also won't be incremented in test DB (SQLite)
|
|
async def delete_object(self, key):
|
|
pass
|
|
|
|
with patch("services.cloud_analysis_processing._classify_cloud_item", new_callable=AsyncMock, return_value=["test-topic"]):
|
|
proc_result = await process_job_item(
|
|
db_session,
|
|
job_id=job_id,
|
|
item_id=item_row.id,
|
|
user_id=user_id,
|
|
connection_id=conn.id,
|
|
cloud_item_id=item.id,
|
|
minio_client=FakeMinIO(),
|
|
provider_adapter=read_only,
|
|
)
|
|
|
|
assert read_only.mutation_call_count == 0, (
|
|
f"Processing must not call provider mutations: "
|
|
f"upload={read_only.upload_calls}, delete={read_only.delete_calls}, "
|
|
f"rename={read_only.rename_calls}, move={read_only.move_calls}, "
|
|
f"create_folder={read_only.create_folder_calls}"
|
|
)
|
|
# Should succeed or fail gracefully (not from a mutation)
|
|
assert proc_result.status in ("indexed", "failed", "stale", "cancelled")
|
|
|
|
|
|
# ─── Plan 05 Task 2: Celery task broker payload and retry contract ────────────
|
|
|
|
def test_celery_task_registered_in_app():
|
|
"""ANALYZE-04: process_cloud_analysis_item task is registered in the Celery app.
|
|
|
|
Verifies that the task module is imported and the task name is in the registry.
|
|
"""
|
|
from celery_app import celery_app
|
|
import tasks.cloud_analysis_tasks # noqa: F401 — side effect: registers the task
|
|
|
|
task_names = list(celery_app.tasks.keys())
|
|
assert "tasks.cloud_analysis_tasks.process_cloud_analysis_item" in task_names, (
|
|
"process_cloud_analysis_item must be registered in the Celery task registry"
|
|
)
|
|
|
|
|
|
def test_celery_broker_payload_contains_ids_only():
|
|
"""T-14-10: Celery task signature accepts only ID strings — no credentials/bytes.
|
|
|
|
Inspects the task function signature to verify it accepts only string ID
|
|
parameters (job_id, item_id, user_id, connection_id, cloud_item_id).
|
|
"""
|
|
import inspect
|
|
from tasks.cloud_analysis_tasks import process_cloud_analysis_item
|
|
|
|
sig = inspect.signature(process_cloud_analysis_item)
|
|
params = list(sig.parameters.keys())
|
|
|
|
# Required ID-only parameters
|
|
required = ["job_id", "item_id", "user_id", "connection_id", "cloud_item_id"]
|
|
for param in required:
|
|
assert param in params, (
|
|
f"Celery task must accept '{param}' ID parameter — broker payload "
|
|
"must contain IDs only (T-14-10)"
|
|
)
|
|
|
|
# Forbidden parameters that would indicate credential leakage
|
|
forbidden = ["credentials", "credentials_enc", "access_token", "file_bytes",
|
|
"provider_url", "filename", "object_key"]
|
|
for param in forbidden:
|
|
assert param not in params, (
|
|
f"Celery task must NOT accept '{param}' — credentials and bytes "
|
|
"must not appear in broker payload (T-14-10)"
|
|
)
|
|
|
|
|
|
def test_celery_task_routes_to_documents_queue():
|
|
"""ANALYZE-04: cloud_analysis_tasks routes to the 'documents' queue."""
|
|
from celery_app import celery_app
|
|
|
|
routes = celery_app.conf.task_routes or {}
|
|
assert "tasks.cloud_analysis_tasks.*" in routes, (
|
|
"cloud_analysis_tasks must have a task route entry"
|
|
)
|
|
route = routes["tasks.cloud_analysis_tasks.*"]
|
|
assert route.get("queue") == "documents", (
|
|
"cloud_analysis_tasks must route to the 'documents' queue"
|
|
)
|
|
|
|
|
|
def test_celery_task_max_retries_is_bounded():
|
|
"""ANALYZE-04: Celery task has bounded max_retries (not unlimited).
|
|
|
|
Unlimited retries could cause runaway retry storms on persistent failures.
|
|
"""
|
|
from tasks.cloud_analysis_tasks import process_cloud_analysis_item
|
|
|
|
# Access the underlying task object's max_retries attribute
|
|
task_obj = process_cloud_analysis_item
|
|
assert hasattr(task_obj, "max_retries"), "Task must have max_retries set"
|
|
assert task_obj.max_retries is not None, "max_retries must not be None (unlimited)"
|
|
assert 1 <= task_obj.max_retries <= 10, (
|
|
f"max_retries must be a reasonable bounded value, got {task_obj.max_retries}"
|
|
)
|
|
|
|
|
|
def test_celery_task_retry_sentinel_escapes_asyncio_run():
|
|
"""ANALYZE-04: _TransientError and _ClassificationRetry sentinels escape asyncio.run().
|
|
|
|
The sentinel pattern requires that exceptions raised inside asyncio.run()
|
|
propagate out to the sync Celery task layer where self.retry() can be called.
|
|
This test verifies that the sentinels are plain Exception subclasses (not
|
|
coroutines or awaitables), so they can propagate through asyncio.run() normally.
|
|
"""
|
|
from tasks.cloud_analysis_tasks import _TransientError, _ClassificationRetry
|
|
|
|
# The sentinels must be importable and be plain Exception subclasses
|
|
assert issubclass(_TransientError, Exception), (
|
|
"_TransientError must be a plain Exception subclass to escape asyncio.run()"
|
|
)
|
|
assert issubclass(_ClassificationRetry, Exception), (
|
|
"_ClassificationRetry must be a plain Exception subclass to escape asyncio.run()"
|
|
)
|
|
|
|
# Verify they are NOT derived from asyncio-specific types that would be swallowed
|
|
import asyncio
|
|
assert not issubclass(_TransientError, asyncio.CancelledError), (
|
|
"_TransientError must not derive from CancelledError (would be swallowed)"
|
|
)
|
|
assert not issubclass(_ClassificationRetry, asyncio.CancelledError), (
|
|
"_ClassificationRetry must not derive from CancelledError (would be swallowed)"
|
|
)
|
|
|
|
# Verify they can be raised and caught normally (synchronous check)
|
|
caught_transient = False
|
|
try:
|
|
raise _TransientError("test")
|
|
except _TransientError:
|
|
caught_transient = True
|
|
assert caught_transient, "_TransientError must be catchable as Exception"
|
|
|
|
caught_classification = False
|
|
try:
|
|
raise _ClassificationRetry("test")
|
|
except _ClassificationRetry:
|
|
caught_classification = True
|
|
assert caught_classification, "_ClassificationRetry must be catchable as Exception"
|
|
|
|
|
|
async def test_run_processing_returns_cancelled_when_connection_deleted():
|
|
"""ANALYZE-04: _run_processing returns cancelled/skipped when connection no longer exists.
|
|
|
|
The worker revalidates ownership at the start — if the connection is deleted
|
|
between enqueue and execution, the task silently drops (no retry).
|
|
|
|
Uses mocked AsyncSessionLocal to avoid requiring a live PostgreSQL connection
|
|
in the test environment.
|
|
"""
|
|
from tasks.cloud_analysis_tasks import _run_processing
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
from services.cloud_items import ConnectionNotFound
|
|
|
|
import uuid
|
|
|
|
# Mock the AsyncSessionLocal so the task uses an in-memory session
|
|
mock_session = AsyncMock()
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
# Mock resolve_owned_connection to raise ConnectionNotFound (deleted connection)
|
|
with patch("tasks.cloud_analysis_tasks._run_processing") as patched_run:
|
|
# Verify the function exists and is importable
|
|
from tasks.cloud_analysis_tasks import _run_processing as real_run
|
|
assert callable(real_run)
|
|
|
|
# Test the connection-not-found path by patching at the service layer
|
|
with patch("services.cloud_items.resolve_owned_connection",
|
|
new_callable=AsyncMock,
|
|
side_effect=ConnectionNotFound("not found")):
|
|
with patch("db.session.AsyncSessionLocal") as mock_session_local:
|
|
mock_ctx = AsyncMock()
|
|
mock_ctx.__aenter__ = AsyncMock(return_value=mock_ctx)
|
|
mock_ctx.__aexit__ = AsyncMock(return_value=False)
|
|
mock_session_local.return_value = mock_ctx
|
|
|
|
result = await _run_processing(
|
|
job_id=str(uuid.uuid4()),
|
|
item_id=str(uuid.uuid4()),
|
|
user_id=str(uuid.uuid4()),
|
|
connection_id=str(uuid.uuid4()), # Does not exist in DB
|
|
cloud_item_id=str(uuid.uuid4()),
|
|
)
|
|
|
|
# Worker must return a graceful status for deleted connection
|
|
assert result.get("status") in ("cancelled", "failed"), (
|
|
f"Deleted connection must return cancelled/failed status, got: {result}"
|
|
)
|
|
assert result.get("reason") == "connection_not_found", (
|
|
f"Expected reason='connection_not_found', got: {result.get('reason')}"
|
|
)
|