diff --git a/.planning/phases/12-cloud-resource-foundation/12-02-SUMMARY.md b/.planning/phases/12-cloud-resource-foundation/12-02-SUMMARY.md new file mode 100644 index 0000000..1be03ae --- /dev/null +++ b/.planning/phases/12-cloud-resource-foundation/12-02-SUMMARY.md @@ -0,0 +1,162 @@ +--- +phase: "12" +plan: "02" +subsystem: cloud-browse +tags: [cloud, browse, celery, idor, stale-while-revalidate, provider-normalization] +dependency_graph: + requires: + - "12-01" # CloudResourceAdapter contract, cloud_base.py, db migrations + provides: + - GET /api/cloud/connections/{connection_id}/items + - PATCH /api/cloud/connections/{id} + - tasks.cloud_tasks.refresh_cloud_folder + - CloudResourceAdapter mixin on all 4 providers + affects: + - backend/api/cloud/ (new package) + - backend/services/cloud_items.py + - backend/tasks/cloud_tasks.py +tech_stack: + added: + - CloudResourceAdapter abstract mixin (storage/cloud_base.py) + - refresh_cloud_folder Celery task (tasks/cloud_tasks.py) + - api/cloud/ package with connections.py, browse.py, schemas.py + patterns: + - stale-while-revalidate (5-min threshold, background Celery refresh) + - sentinel exception pattern for Celery retry in sync layer + - whitelisted Pydantic response schemas (credential-free) + - UUID coerce-to-uuid.UUID pattern for SQLAlchemy UUID(as_uuid=True) columns +key_files: + created: + - backend/api/cloud/__init__.py + - backend/api/cloud/browse.py + - backend/api/cloud/connections.py + - backend/api/cloud/schemas.py + - backend/tasks/cloud_tasks.py + - AGENTS.md + modified: + - backend/storage/google_drive_backend.py + - backend/storage/onedrive_backend.py + - backend/storage/webdav_backend.py + - backend/storage/nextcloud_backend.py + - backend/storage/cloud_backend_factory.py + - backend/services/cloud_items.py + - backend/db/models.py + - backend/celery_app.py + - backend/main.py + - frontend/package.json + - backend/tests/test_cloud.py + - backend/tests/test_cloud_items.py + - backend/tests/conftest.py + - README.md +decisions: + - "UUID coercion in cloud_items.py service layer: always converts to uuid.UUID objects for WHERE clause binding with UUID(as_uuid=True) columns; test_cloud_items.py updated to use UUID(as_uuid=True) instead of String(36) patch to match production ORM" + - "Stale-while-revalidate threshold set to 5 minutes: returns cached rows immediately on first browse if stale, schedules background refresh" + - "refresh_cloud_folder: decrypts credentials inside worker, never in broker payload; 3-retry bounded backoff 30s/90s/270s with ±10s jitter" + - "Celery retry sentinel pattern: _TransientProviderError and _TerminalProviderError escape asyncio.run() to Celery sync layer" + - "display_name_override ORM field added to CloudConnection model (migration 0006 already had the column)" +metrics: + duration: "~4 hours (continued from prior session)" + completed: "2026-06-18T21:17:38Z" + tasks_completed: 3 + files_changed: 19 +--- + +# Phase 12 Plan 02: Cloud Resource Foundation Browse API Summary + +Connection-ID browse endpoint with all four providers implementing `CloudResourceAdapter`, stale-while-revalidate caching, and durable `refresh_cloud_folder` Celery task. + +## What Was Built + +**Task 1: Provider normalization (committed ff33439)** +All four cloud providers (Google Drive, OneDrive, WebDAV, Nextcloud) now implement `CloudResourceAdapter` as a mixin. Each provider adds `list_folder()` and `get_capabilities()` returning normalized `CloudListing` and capability dicts. Factory adds `build_cloud_resource_adapter()`. 22 new tests in `test_cloud_backends.py` verify pagination, no-byte-download, and capability evidence contracts. + +**Task 2: api/cloud/ package decomposition (committed e186019)** +Replaced flat `api/cloud.py` with a package: +- `connections.py`: all connection management routes (OAuth, WebDAV, list, delete, rename) +- `browse.py`: canonical `GET /api/cloud/connections/{connection_id}/items` +- `schemas.py`: credential-free `CloudBrowseResponse`, `CloudItemOut`, `FolderFreshnessOut` +- `__init__.py`: aggregates sub-routers under `/api/cloud` prefix + +Fixed UUID type incompatibility: `test_cloud_items.py` now uses `UUID(as_uuid=True)` matching the production ORM instead of `String(36)` patch. Service layer coerces all inputs to `uuid.UUID` objects for WHERE clauses. + +**Task 3: Celery task + version bump (committed c6237ca)** +- `refresh_cloud_folder` Celery task: idempotent, bounded 3-retry (30s/90s/270s ± 10s jitter), credential decryption inside worker +- Stale-while-revalidate: 5-minute threshold in browse endpoint +- Version bumped 0.1.4 → 0.1.5 in backend/main.py and frontend/package.json +- AGENTS.md created/updated with Phase 12 state and new shared module map entries +- README.md updated with connection-ID browse API table + +## Commits + +| Hash | Message | +|------|---------| +| ff33439 | feat(12-02): normalize all four providers into CloudResourceAdapter contract | +| e186019 | feat(12-02): decompose api/cloud.py into package with connection-ID browse endpoint | +| c6237ca | feat(12-02): add refresh_cloud_folder Celery task, staleness trigger, version 0.1.5 | + +## Test Results + +493 passing, 1 pre-existing failure (`test_extract_docx` — missing `python-docx` module, unrelated to this plan), 6 skipped, 7 xfailed. + +New tests added: +- 22 in `test_cloud_backends.py` (Task 1) +- 11 in `test_cloud.py` (Task 2: IDOR, admin block, credential exclusion, rename, duplicate providers, malformed UUID) +- 5 in `test_cloud_items.py` (Task 3: idempotency, cached-row retention, task structure, no-byte-download, scheduling) + +## Deviations from Plan + +### Auto-fixed Issues + +**1. [Rule 1 - Bug] UUID type incompatibility between test fixtures** +- **Found during:** Task 2 debugging +- **Issue:** `test_cloud_items.py` patched UUID columns to `String(36)` while conftest used `UUID(as_uuid=True)`. Service functions could not work with both simultaneously — UUID objects caused InterfaceError on String(36) columns; strings caused AttributeError (.hex) on UUID(as_uuid=True) columns. +- **Fix:** Updated `test_cloud_items.py` fixture to use `UUID(as_uuid=True)` approach (matching production ORM). Updated all helpers to use `uuid.UUID` objects. Updated `cloud_items.py` service to always coerce to `uuid.UUID` objects in WHERE clauses. +- **Files modified:** `backend/tests/test_cloud_items.py`, `backend/services/cloud_items.py`, `backend/tests/conftest.py` +- **Commits:** e186019, c6237ca + +**2. [Rule 2 - Missing functionality] display_name_override missing from ORM model** +- **Found during:** Task 2 +- **Issue:** Migration 0006 added `display_name_override` column to `cloud_connections` table but ORM model didn't declare it, causing AttributeError when PATCH endpoint tried to set it. +- **Fix:** Added `display_name_override: Mapped[Optional[str]] = mapped_column(Text, nullable=True)` to `CloudConnection` model. +- **Files modified:** `backend/db/models.py` +- **Commit:** e186019 + +**3. [Rule 1 - Bug] Celery patch target wrong in refresh scheduling test** +- **Found during:** Task 3 test authoring +- **Issue:** Test patched `api.cloud.browse.refresh_cloud_folder` but browse.py uses a local import inside the function body, so the module attribute doesn't exist at test time. +- **Fix:** Changed patch target to `tasks.cloud_tasks.refresh_cloud_folder`. +- **Files modified:** `backend/tests/test_cloud.py` +- **Commit:** c6237ca + +**4. [Rule 1 - Bug] should_refresh condition never triggered for stale cached items** +- **Found during:** Task 3 scheduling test +- **Issue:** Browse endpoint's `should_refresh` only checked for no items, "refreshing" state, or `last_refreshed_at is None`. Cached items with `refresh_state="fresh"` and a recent `last_refreshed_at` would never schedule background refresh. +- **Fix:** Added 5-minute staleness check — if `last_refreshed_at` is older than 5 minutes, refresh is scheduled. +- **Files modified:** `backend/api/cloud/browse.py` +- **Commit:** c6237ca + +## Known Stubs + +None — all data sources wired. Browse endpoint returns real durable cached rows from `cloud_items` table and schedules real Celery task. + +## Threat Flags + +| Flag | File | Description | +|------|------|-------------| +| threat_flag: credential_in_worker | backend/tasks/cloud_tasks.py | Credentials decrypted inside Celery worker — master key must be in CLOUD_CREDS_KEY env var on all worker nodes. Never log credentials. Already mitigated by existing key-from-env pattern. | + +## Self-Check: PASSED + +Files confirmed created: +- backend/api/cloud/__init__.py ✓ +- backend/api/cloud/browse.py ✓ +- backend/api/cloud/connections.py ✓ +- backend/api/cloud/schemas.py ✓ +- backend/tasks/cloud_tasks.py ✓ + +Commits confirmed: +- ff33439 ✓ +- e186019 ✓ +- c6237ca ✓ + +Test result: 493 passed, 1 pre-existing failure ✓ diff --git a/AGENTS.md b/AGENTS.md index f9822cf..12f961f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -4,7 +4,7 @@ DocuVault is a multi-user SaaS document management platform built on FastAPI (Python) + Vue 3. It handles document upload, text extraction (PDF/DOCX/image/text), AI-based topic classification, per-user isolated storage, folder organization, document sharing, and pluggable cloud storage backends (OneDrive, Google Drive, Nextcloud, WebDAV). -**Current state:** v0.1 alpha — not production-ready. Multi-user SaaS with full auth, PostgreSQL + MinIO, cloud storage backends (OneDrive, Google Drive, Nextcloud, WebDAV), folder/share/quota management, structured observability, and a refactored AI provider layer backed by a `system_settings` DB table. All 7 phases complete as of 2026-06-05. The app is functional for local/self-hosted use but has not been hardened or audited for public internet deployment. +**Current state:** v0.1.5 — Phase 12 Plan 02 complete (2026-06-18). Connection-ID browse API (`GET /api/cloud/connections/{connection_id}/items`) with owner-scoped IDOR protection, stale-while-revalidate caching, and durable `refresh_cloud_folder` Celery task. All 4 providers implement `CloudResourceAdapter`. Not cleared for public internet deployment. ## Stack @@ -23,6 +23,8 @@ DocuVault is a multi-user SaaS document management platform built on FastAPI (Py - Admin endpoints **never return** document content, extracted text, or `credentials_enc` - Every document/folder endpoint asserts `resource.user_id == current_user.id` - All DB queries via ORM / parameterized statements — zero raw string interpolation +- Cloud browse/refresh **never downloads file bytes** and never mutates quota (D-18/CACHE-01) +- `refresh_cloud_folder` Celery task decrypts credentials inside the worker — never in broker payload ## Code Standards (Non-Negotiable) @@ -40,6 +42,9 @@ Before adding a helper, check if it belongs in an existing shared module: | `backend/storage/exceptions.py` | `CloudConnectionError` — single canonical definition; all files import from here | | `backend/ai/utils.py` | `strip_code_fences`, `parse_classification`, `parse_suggestions` — AI response parsing shared by all providers | | `backend/services/auth.py` | `validate_password_strength(password)` — raises `ValueError`; routers catch and re-raise as `HTTPException` | +| `backend/storage/cloud_base.py` | `CloudResourceAdapter`, `CloudListing`, `CloudResource`, `CloudCapability` — Phase 12 browse contract | +| `backend/services/cloud_items.py` | `resolve_owned_connection`, `upsert_cloud_item`, `reconcile_cloud_listing` — owner-scoped metadata service | +| `backend/api/cloud/schemas.py` | `CloudBrowseResponse`, `CloudItemOut`, `FolderFreshnessOut` — credential-free response schemas | **Rules:** - No router may define `_ip()`, `_get_ip()`, or any other local variant of `get_client_ip`. Import from `deps.utils`. @@ -47,6 +52,8 @@ Before adding a helper, check if it belongs in an existing shared module: - No AI provider may define its own `_strip_code_fences` or `_parse_*`. Import from `ai.utils`. - No API file may define `_validate_password_strength`. Import from `services.auth`. - Service layer raises `ValueError` (or domain exceptions), never `HTTPException`. Only the router layer raises `HTTPException`. +- Cloud browse responses must use `api/cloud/schemas.py` whitelisted types — never return raw ORM objects. +- `reconcile_cloud_listing` is the single reconciliation entry point — no provider may update `cloud_items` rows directly. ### Frontend: shared module map @@ -116,9 +123,9 @@ This project uses the GSD (Get Shit Done) planning workflow. Planning artifacts /gsd:progress — check status and advance workflow ``` -### Current state: v0.1 alpha — Phase 11 mobile storage UAT gap closure complete (2026-06-17) +### Current state: v0.1.5 — Phase 12 Plan 02 complete (2026-06-18) -Phase 11 now includes the 11-07 UAT gap closure for touch-safe storage row actions and compact mobile search/sort/new-folder controls. The app is functional but alpha-quality — not cleared for public internet deployment. The next milestone has not been started. +Phase 12 Plan 02: connection-ID browse endpoint, `api/cloud/` package decomposition, `CloudResourceAdapter` mixin on all 4 providers, and `refresh_cloud_folder` Celery task with bounded retry. Stale-while-revalidate caching active. The app is functional but alpha-quality — not cleared for public internet deployment. ## Development Setup diff --git a/README.md b/README.md index 5c15d4d..46946ac 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # DocuVault -**Version 0.1.4 — Alpha** +**Version 0.1.5 — Alpha** > **Not production-ready.** DocuVault is functional for local and self-hosted use but has not been audited or hardened for public internet exposure. APIs, environment variables, and the database schema may change without notice until a stable 1.0 release is declared. @@ -39,7 +39,7 @@ A self-hosted, multi-user document management platform with AI-powered topic cla │ ├── api/documents.py │ │ ├── api/folders.py │ │ ├── api/shares.py │ -│ ├── api/cloud.py (cloud backends) │ +│ ├── api/cloud/ (cloud backends package) │ │ ├── api/admin.py │ │ └── api/audit.py │ └──┬──────────┬──────────┬───────────────┘ @@ -303,6 +303,19 @@ Users connect cloud storage through **Settings → Cloud Storage**. Credentials Connection statuses: `ACTIVE`, `REQUIRES_REAUTH`, `ERROR`. An externally revoked OAuth token transitions to `REQUIRES_REAUTH` without a 500 error. +### Connection-ID Browse API (Phase 12) + +Each connected account is independently addressable by its connection UUID: + +| Endpoint | Purpose | +|----------|---------| +| `GET /api/cloud/connections` | List all connections (all providers, no credentials) | +| `GET /api/cloud/connections/{id}/items` | Browse folder by connection UUID (stale-while-revalidate) | +| `PATCH /api/cloud/connections/{id}` | Rename connection display name | +| `DELETE /api/cloud/connections/{id}` | Remove connection and credentials | + +Browsing returns durable cached rows immediately and schedules a background `refresh_cloud_folder` Celery task to reconcile provider changes. First-visit fetches are synchronous and bounded. All browse responses are credential-free — `credentials_enc` is never serialized in the response. + --- ## Security Highlights diff --git a/backend/api/cloud/__init__.py b/backend/api/cloud/__init__.py new file mode 100644 index 0000000..f03cd22 --- /dev/null +++ b/backend/api/cloud/__init__.py @@ -0,0 +1,63 @@ +""" +Cloud API package — router aggregator. + +One parent router owns the /api/cloud prefix. +Sub-routers (connections, browse) carry no prefix — paths are declared +relative to the parent (D-04 pattern, same as api/documents/__init__.py). + +main.py imports: + from api.cloud import router as cloud_router, users_router + app.include_router(cloud_router) + app.include_router(users_router) +""" +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel +from sqlalchemy.ext.asyncio import AsyncSession + +from api.cloud.connections import router as connections_router, _VALID_BACKENDS +from api.cloud.browse import router as browse_router +from db.models import User +from deps.auth import get_regular_user +from deps.db import get_db +from services.rate_limiting import account_limiter + +router = APIRouter(prefix="/api/cloud", tags=["cloud"]) +router.include_router(connections_router) +router.include_router(browse_router) + +# ── Users router (default storage) ─────────────────────────────────────────── + +users_router = APIRouter(prefix="/api/users", tags=["users"]) + + +class DefaultStorageRequest(BaseModel): + backend: str + + +@users_router.patch("/me/default-storage") +@account_limiter.limit("100/minute") +async def update_default_storage( + request: Request, + body: DefaultStorageRequest, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> dict: + """Update the current user's default storage backend. + + Validated against the allowlist before storage. + """ + if body.backend not in _VALID_BACKENDS: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"Invalid backend. Valid values: {sorted(_VALID_BACKENDS)}", + ) + request.state.current_user = current_user + user = await session.get(User, current_user.id) + if user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") + + user.default_storage_backend = body.backend + session.add(user) + await session.commit() + + return {"default_storage_backend": user.default_storage_backend} diff --git a/backend/api/cloud/browse.py b/backend/api/cloud/browse.py new file mode 100644 index 0000000..bf76e8d --- /dev/null +++ b/backend/api/cloud/browse.py @@ -0,0 +1,340 @@ +""" +Canonical connection-ID browse endpoint for Phase 12. + +GET /api/cloud/connections/{connection_id}/items + - Owner-scoped: resolves connection by UUID, not provider name + - Stale-while-revalidate: returns durable cached rows immediately; schedules + background refresh if stale or first visit + - T-12-01: IDOR-protected via resolve_owned_connection + - T-12-03: response uses whitelisted schemas, no credentials in response + - D-18/CACHE-01: never downloads file bytes, never mutates quota +""" +from __future__ import annotations + +import asyncio +import uuid +from typing import Optional + +import httpx +from fastapi import APIRouter, Depends, HTTPException, Request, status +from sqlalchemy.ext.asyncio import AsyncSession + +from api.cloud.schemas import ( + CloudBrowseResponse, + CloudCapabilityOut, + CloudItemOut, + FolderFreshnessOut, +) +from db.models import CloudConnection, CloudItem, CloudFolderState, User +from deps.auth import get_regular_user +from deps.db import get_db +from deps.utils import parse_uuid +from services.cloud_items import ( + ConnectionNotFound, + list_cloud_children, + get_or_create_folder_state, + reconcile_cloud_listing, + resolve_owned_connection, + update_folder_state, +) +from services.rate_limiting import account_limiter +from storage.cloud_backend_factory import build_cloud_resource_adapter +from storage.cloud_utils import decrypt_credentials, encrypt_credentials +from config import settings + +router = APIRouter() + +_DISPLAY_NAMES = { + "google_drive": "Google Drive", + "onedrive": "OneDrive", + "nextcloud": "Nextcloud", + "webdav": "WebDAV server", +} + + +def _master_key() -> bytes: + return settings.cloud_creds_key.encode() + + +def _decrypt_connection(conn: CloudConnection, user_id: uuid.UUID) -> dict: + return decrypt_credentials(_master_key(), str(user_id), conn.credentials_enc) + + +def _capability_out(caps: dict) -> dict[str, CloudCapabilityOut]: + return { + action: CloudCapabilityOut( + action=cap.action, + state=cap.state, + reason=cap.reason, + message=cap.message, + ) + for action, cap in caps.items() + } + + +def _item_out(item: CloudItem) -> CloudItemOut: + return CloudItemOut( + id=str(item.id), + provider_item_id=item.provider_item_id, + name=item.name, + kind=item.kind, + parent_ref=item.parent_ref, + content_type=item.content_type, + size=item.provider_size, + modified_at=item.modified_at, + etag=item.etag, + capabilities={}, # Phase 13 will add per-item capabilities + ) + + +def _freshness_out(fs: CloudFolderState) -> FolderFreshnessOut: + return FolderFreshnessOut( + refresh_state=fs.refresh_state, + last_refreshed_at=fs.last_refreshed_at, + error_code=fs.error_code, + error_message=fs.error_message, + ) + + +# ── Legacy helpers used by compatibility route in connections.py ───────────── + +async def _fetch_google_drive_folders(credentials: dict, folder_id: str) -> list: + """Legacy listing helper preserved for compatibility route.""" + import datetime as dt + from google.oauth2.credentials import Credentials as GoogleCreds + from googleapiclient.discovery import build + + expiry = None + if expiry_str := credentials.get("expiry"): + try: + expiry = dt.datetime.fromisoformat(expiry_str) + except ValueError: + pass + + creds = GoogleCreds( + token=credentials.get("access_token"), + refresh_token=credentials.get("refresh_token"), + token_uri=credentials.get("token_uri", "https://oauth2.googleapis.com/token"), + client_id=credentials.get("client_id"), + client_secret=credentials.get("client_secret"), + expiry=expiry, + ) + + def _list_files() -> list: + service = build("drive", "v3", credentials=creds, cache_discovery=False) + response = service.files().list( + q=f"'{folder_id}' in parents and trashed=false", + fields="files(id,name,mimeType,size)", + pageSize=200, + ).execute() + items = [] + for item in response.get("files", []): + is_dir = item.get("mimeType") == "application/vnd.google-apps.folder" + items.append({ + "id": item["id"], + "name": item["name"], + "is_dir": is_dir, + "size": int(item.get("size", 0)) if not is_dir else 0, + }) + return items + + return await asyncio.to_thread(_list_files) + + +async def _fetch_onedrive_folders(credentials: dict, folder_id: str) -> list: + """Legacy listing helper preserved for compatibility route.""" + if folder_id in ("root", ""): + url = "https://graph.microsoft.com/v1.0/me/drive/root/children" + else: + url = f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_id}/children" + + async with httpx.AsyncClient() as client: + resp = await client.get( + url, + headers={"Authorization": f"Bearer {credentials.get('access_token', '')}"}, + timeout=30, + ) + resp.raise_for_status() + data = resp.json() + + items = [] + for item in data.get("value", []): + is_dir = "folder" in item + items.append({ + "id": item["id"], + "name": item["name"], + "is_dir": is_dir, + "size": item.get("size", 0) if not is_dir else 0, + }) + return items + + +async def _fetch_webdav_folders(provider: str, credentials: dict, folder_id: str) -> list: + """Legacy listing helper preserved for compatibility route.""" + from storage.cloud_backend_factory import build_cloud_backend + webdav_path = "" if folder_id == "root" else folder_id + backend = build_cloud_backend(provider, credentials) + return await backend.list_folder(webdav_path) + + +# ── Canonical connection-ID browse endpoint ────────────────────────────────── + +@router.get("/connections/{connection_id}/items", response_model=CloudBrowseResponse) +@account_limiter.limit("100/minute") +async def browse_connection_items( + connection_id: uuid.UUID, + request: Request, + parent_ref: Optional[str] = None, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> CloudBrowseResponse: + """Browse the items of a connection by UUID. + + T-12-01: IDOR protection — connection resolved by owner UUID, not provider alone. + T-12-03: response schema excludes credentials_enc and all credential fields. + D-18/CACHE-01: returns durable cached rows; never downloads file bytes. + D-11/D-12: cached rows returned immediately; background refresh scheduled if stale. + D-13/D-14: on refresh failure, cached rows retained with warning state. + """ + request.state.current_user = current_user + + uid_str = str(current_user.id) + cid_str = str(connection_id) + + # Resolve connection ownership (T-12-01) + try: + conn = await resolve_owned_connection( + session, + connection_id=cid_str, + user_id=uid_str, + ) + except ConnectionNotFound: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found") + + # Ensure folder state row exists + folder_state = await get_or_create_folder_state( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref or "", + ) + await session.commit() + + # Return durable cached rows + cached_items = await list_cloud_children( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref, + ) + + # Get connection-level capabilities (from a fresh call to avoid stale state) + try: + credentials = _decrypt_connection(conn, current_user.id) + adapter = build_cloud_resource_adapter(conn.provider, credentials) + conn_caps = await adapter.get_capabilities(connection_id, current_user.id) + except Exception: + from storage.cloud_base import ACTIONS, CloudCapability, STATE_TEMPORARILY_UNAVAILABLE, REASON_OFFLINE + conn_caps = { + action: CloudCapability( + action=action, + state=STATE_TEMPORARILY_UNAVAILABLE, + reason=REASON_OFFLINE, + message="Provider temporarily unreachable.", + ) + for action in ACTIONS + } + + # Refresh in background if first visit (no items), currently refreshing, + # never refreshed, or stale (last refresh > 5 minutes ago) + import datetime as _dt + _stale_threshold = _dt.timedelta(minutes=5) + _is_stale = ( + folder_state.last_refreshed_at is None + or (_dt.datetime.now(_dt.timezone.utc) - folder_state.last_refreshed_at) > _stale_threshold + ) + should_refresh = ( + not cached_items + or folder_state.refresh_state in ("refreshing",) + or _is_stale + ) + if should_refresh and not cached_items: + # First visit: do a synchronous bounded fetch so user sees content + try: + credentials = _decrypt_connection(conn, current_user.id) + adapter = build_cloud_resource_adapter(conn.provider, credentials) + listing = await adapter.list_folder( + connection_id, + current_user.id, + parent_ref=parent_ref, + ) + await reconcile_cloud_listing( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref, + listing=listing, + ) + await update_folder_state( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref or "", + refresh_state="fresh", + ) + await session.commit() + # Re-read items after reconciliation + cached_items = await list_cloud_children( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref, + ) + # Re-fetch folder state + folder_state = await get_or_create_folder_state( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref or "", + ) + except Exception as exc: + # Retain cached rows (empty on first visit), set warning state + await update_folder_state( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref or "", + refresh_state="warning", + error_code="provider_error", + error_message="Provider temporarily unavailable. Retrying in background.", + ) + await session.commit() + folder_state = await get_or_create_folder_state( + session, + user_id=uid_str, + connection_id=cid_str, + parent_ref=parent_ref or "", + ) + elif should_refresh and cached_items: + # Has cached items — schedule background refresh via Celery + try: + from tasks.cloud_tasks import refresh_cloud_folder + refresh_cloud_folder.delay( + str(current_user.id), + str(connection_id), + parent_ref, + ) + except Exception: + pass # Celery unavailable — serve stale cache without failing browse + + display_name = conn.display_name_override or conn.display_name + + return CloudBrowseResponse( + connection_id=str(connection_id), + provider=conn.provider, + display_name=display_name, + parent_ref=parent_ref, + items=[_item_out(item) for item in cached_items], + capabilities=_capability_out(conn_caps), + freshness=_freshness_out(folder_state), + ) diff --git a/backend/api/cloud.py b/backend/api/cloud/connections.py similarity index 83% rename from backend/api/cloud.py rename to backend/api/cloud/connections.py index b2d1a33..faa6bba 100644 --- a/backend/api/cloud.py +++ b/backend/api/cloud/connections.py @@ -1,4 +1,9 @@ -"""Cloud storage connection management endpoints.""" +""" +Connection management endpoints — all existing cloud connection behaviors. + +Preserves all existing endpoint URLs from api/cloud.py without change. +Introduces PATCH /api/cloud/connections/{connection_id} for display_name rename. +""" from __future__ import annotations import asyncio @@ -13,19 +18,19 @@ from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from api.cloud.schemas import ConnectionRenameRequest from api.schemas import CloudConnectionOut from config import settings from db.models import CloudConnection, User from deps.auth import get_regular_user from deps.db import get_db -from deps.utils import get_client_ip +from deps.utils import get_client_ip, parse_uuid from services.audit import write_audit_log from services.rate_limiting import account_limiter from storage.cloud_backend_factory import build_cloud_backend from storage.cloud_utils import decrypt_credentials, encrypt_credentials, validate_cloud_url -router = APIRouter(prefix="/api/cloud", tags=["cloud"]) -users_router = APIRouter(prefix="/api/users", tags=["users"]) +router = APIRouter() VALID_OAUTH_PROVIDERS = {"google_drive", "onedrive"} VALID_WEBDAV_PROVIDERS = {"nextcloud", "webdav"} @@ -107,6 +112,7 @@ async def _upsert_cloud_connection( provider: str, credentials_enc: str, ) -> CloudConnection: + """Insert or update a connection for (user_id, provider).""" result = await session.execute( select(CloudConnection).where( CloudConnection.user_id == user_id, @@ -137,6 +143,7 @@ async def _get_owned_connection( connection_id: uuid.UUID, user_id: uuid.UUID, ) -> CloudConnection: + """Return the connection owned by user_id, or raise 404.""" conn = await session.get(CloudConnection, connection_id) if conn is None or conn.user_id != user_id: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Connection not found") @@ -278,83 +285,7 @@ def _connection_response(conn: CloudConnection) -> dict: return data -async def _fetch_google_drive_folders(credentials: dict, folder_id: str) -> list: - from google.oauth2.credentials import Credentials # lazy import - from googleapiclient.discovery import build # lazy import - - import datetime as dt - - expiry = None - if expiry_str := credentials.get("expiry"): - try: - expiry = dt.datetime.fromisoformat(expiry_str) - except ValueError: - pass - - creds = Credentials( - token=credentials.get("access_token"), - refresh_token=credentials.get("refresh_token"), - token_uri=credentials.get("token_uri", "https://oauth2.googleapis.com/token"), - client_id=credentials.get("client_id"), - client_secret=credentials.get("client_secret"), - expiry=expiry, - ) - - def _list_files() -> list: - service = build("drive", "v3", credentials=creds, cache_discovery=False) - response = service.files().list( - q=f"'{folder_id}' in parents and trashed=false", - fields="files(id,name,mimeType,size)", - pageSize=200, - ).execute() - items = [] - for item in response.get("files", []): - is_dir = item.get("mimeType") == "application/vnd.google-apps.folder" - items.append({ - "id": item["id"], - "name": item["name"], - "is_dir": is_dir, - "size": int(item.get("size", 0)) if not is_dir else 0, - }) - return items - - return await asyncio.to_thread(_list_files) - - -async def _fetch_onedrive_folders(credentials: dict, folder_id: str) -> list: - import httpx # lazy import - - if folder_id in ("root", ""): - url = "https://graph.microsoft.com/v1.0/me/drive/root/children" - else: - url = f"https://graph.microsoft.com/v1.0/me/drive/items/{folder_id}/children" - - async with httpx.AsyncClient() as client: - resp = await client.get( - url, - headers={"Authorization": f"Bearer {credentials.get('access_token', '')}"}, - timeout=30, - ) - resp.raise_for_status() - data = resp.json() - - items = [] - for item in data.get("value", []): - is_dir = "folder" in item - items.append({ - "id": item["id"], - "name": item["name"], - "is_dir": is_dir, - "size": item.get("size", 0) if not is_dir else 0, - }) - return items - - -async def _fetch_webdav_folders(provider: str, credentials: dict, folder_id: str) -> list: - webdav_path = "" if folder_id == "root" else folder_id - backend = build_cloud_backend(provider, credentials) - return await backend.list_folder(webdav_path) - +# ── Endpoints ───────────────────────────────────────────────────────────────── @router.get("/oauth/initiate/{provider}") @account_limiter.limit("100/minute") @@ -561,6 +492,28 @@ async def get_connection_config( } +@router.patch("/connections/{connection_id}", status_code=status.HTTP_200_OK) +@account_limiter.limit("100/minute") +async def rename_connection( + connection_id: uuid.UUID, + body: ConnectionRenameRequest, + request: Request, + session: AsyncSession = Depends(get_db), + current_user: User = Depends(get_regular_user), +) -> dict: + """Rename a cloud connection's display name. + + Only the display_name field is accepted — mass assignment prevention. + T-12-01: owner check via _get_owned_connection. + """ + request.state.current_user = current_user + conn = await _get_owned_connection(session, connection_id, current_user.id) + conn.display_name_override = body.display_name + await session.commit() + await session.refresh(conn) + return {"id": str(conn.id), "display_name": body.display_name} + + @router.delete("/connections/{connection_id}", status_code=status.HTTP_204_NO_CONTENT) @account_limiter.limit("100/minute") async def delete_connection( @@ -604,7 +557,11 @@ async def list_cloud_folders( session: AsyncSession = Depends(get_db), current_user: User = Depends(get_regular_user), ) -> dict: - """List folder contents for a connected cloud provider.""" + """List folder contents for a connected cloud provider (compatibility route). + + This endpoint is the original provider-keyed browse route preserved for + backward compatibility. The canonical route is GET /connections/{id}/items. + """ request.state.current_user = current_user if provider not in VALID_CLOUD_PROVIDERS: raise _unsupported_provider_error(provider, VALID_CLOUD_PROVIDERS) @@ -615,44 +572,17 @@ async def list_cloud_folders( from services.cloud_cache import get_cloud_folders_cached # lazy import if provider == "google_drive": + from api.cloud.browse import _fetch_google_drive_folders fetcher = lambda: _fetch_google_drive_folders(credentials, folder_id) elif provider == "onedrive": + from api.cloud.browse import _fetch_onedrive_folders fetcher = lambda: _fetch_onedrive_folders(credentials, folder_id) else: + from api.cloud.browse import _fetch_webdav_folders fetcher = lambda: _fetch_webdav_folders(provider, credentials, folder_id) items = await get_cloud_folders_cached(str(current_user.id), provider, folder_id, fetcher) return {"items": items} - - -@users_router.patch("/me/default-storage") -@account_limiter.limit("100/minute") -async def update_default_storage( - request: Request, - body: DefaultStorageRequest, - session: AsyncSession = Depends(get_db), - current_user: User = Depends(get_regular_user), -) -> dict: - """Update the current user's default storage backend. - - The backend value is validated against the allowlist before storage. - Returns the updated default_storage_backend value. - """ - if body.backend not in _VALID_BACKENDS: - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, - detail=f"Invalid backend. Valid values: {sorted(_VALID_BACKENDS)}", - ) - request.state.current_user = current_user - user = await session.get(User, current_user.id) - if user is None: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") - - user.default_storage_backend = body.backend - session.add(user) - await session.commit() - - return {"default_storage_backend": user.default_storage_backend} diff --git a/backend/api/cloud/schemas.py b/backend/api/cloud/schemas.py new file mode 100644 index 0000000..1066321 --- /dev/null +++ b/backend/api/cloud/schemas.py @@ -0,0 +1,86 @@ +""" +Whitelisted Pydantic response schemas for the cloud API package. + +All schemas are explicit allowlists — credentials_enc, tokens, and passwords +are deliberately absent. T-12-03: credential exclusion by design. +""" +from __future__ import annotations + +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, field_validator + + +# ── Capability / item schemas ───────────────────────────────────────────────── + +class CloudCapabilityOut(BaseModel): + """Whitelisted capability descriptor. reason/message only when not supported.""" + + action: str + state: str # "supported" | "unsupported" | "temporarily_unavailable" + reason: Optional[str] = None + message: Optional[str] = None + + +class CloudItemOut(BaseModel): + """Normalized cloud item metadata. No credentials or byte content.""" + + id: str # DocuVault stable UUID + provider_item_id: str + name: str + kind: str # "file" | "folder" + parent_ref: Optional[str] = None + content_type: Optional[str] = None + size: Optional[int] = None + modified_at: Optional[datetime] = None + etag: Optional[str] = None + capabilities: dict[str, CloudCapabilityOut] = {} + + +# ── Freshness / folder state schemas ───────────────────────────────────────── + +class FolderFreshnessOut(BaseModel): + """Freshness and error state for a browsed folder.""" + + refresh_state: str # "fresh" | "refreshing" | "warning" + last_refreshed_at: Optional[datetime] = None + error_code: Optional[str] = None + error_message: Optional[str] = None + + +# ── Browse response ─────────────────────────────────────────────────────────── + +class CloudBrowseResponse(BaseModel): + """Owner-scoped connection-ID browse response. + + T-12-01: items are always scoped to the resolved connection which is owned + by the requesting user. credentials_enc is never included. + """ + + connection_id: str + provider: str + display_name: str + parent_ref: Optional[str] + items: List[CloudItemOut] + capabilities: dict[str, CloudCapabilityOut] + freshness: FolderFreshnessOut + + +# ── Connection schemas ──────────────────────────────────────────────────────── + +class ConnectionRenameRequest(BaseModel): + """Validated PATCH body for renaming a connection display name. + + Only display_name is accepted — mass assignment prevention. + """ + + display_name: str + + @field_validator("display_name") + @classmethod + def must_be_nonblank(cls, v: str) -> str: + stripped = v.strip() + if not stripped: + raise ValueError("display_name must not be blank") + return stripped diff --git a/backend/celery_app.py b/backend/celery_app.py index bff1900..5d4b8ef 100644 --- a/backend/celery_app.py +++ b/backend/celery_app.py @@ -35,6 +35,7 @@ celery_app.conf.task_routes = { "tasks.document_tasks.*": {"queue": "documents"}, "tasks.email_tasks.*": {"queue": "email"}, "tasks.audit_tasks.*": {"queue": "documents"}, + "tasks.cloud_tasks.*": {"queue": "documents"}, } # Celery beat schedule: @@ -55,5 +56,6 @@ celery_app.conf.timezone = "UTC" # Explicitly import task modules — autodiscover_tasks(["tasks"]) looks for # tasks.tasks (appends ".tasks") which doesn't exist in our structure. import tasks.audit_tasks # noqa: F401, E402 +import tasks.cloud_tasks # noqa: F401, E402 import tasks.document_tasks # noqa: F401, E402 import tasks.email_tasks # noqa: F401, E402 diff --git a/backend/db/models.py b/backend/db/models.py index 529e36f..e196325 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -309,6 +309,7 @@ class CloudConnection(Base): ) provider: Mapped[str] = mapped_column(String, nullable=False) display_name: Mapped[str] = mapped_column(Text, nullable=False) + display_name_override: Mapped[Optional[str]] = mapped_column(Text, nullable=True) credentials_enc: Mapped[str] = mapped_column(Text, nullable=False) status: Mapped[str] = mapped_column(String, nullable=False, default="ACTIVE") connected_at: Mapped[datetime] = mapped_column( diff --git a/backend/main.py b/backend/main.py index 605af37..afe566f 100644 --- a/backend/main.py +++ b/backend/main.py @@ -244,7 +244,7 @@ async def lifespan(app: FastAPI): # ── Application factory ─────────────────────────────────────────────────────── -app = FastAPI(title="Document Scanner API", version="0.1.4", lifespan=lifespan) +app = FastAPI(title="Document Scanner API", version="0.1.5", lifespan=lifespan) # Rate limiter state (slowapi) app.state.limiter = auth_limiter diff --git a/backend/services/cloud_items.py b/backend/services/cloud_items.py index 02e2df1..c8467cf 100644 --- a/backend/services/cloud_items.py +++ b/backend/services/cloud_items.py @@ -37,17 +37,20 @@ class CloudItemNotFound(ValueError): async def resolve_owned_connection( session: AsyncSession, *, - connection_id: str, - user_id: str, + connection_id, + user_id, ) -> CloudConnection: """Return the CloudConnection owned by user_id or raise ConnectionNotFound. + Accepts UUID objects or string UUIDs for both parameters. Never returns a connection belonging to another user. """ + conn_uuid = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id)) + user_uuid = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) result = await session.execute( select(CloudConnection).where( - CloudConnection.id == connection_id, - CloudConnection.user_id == user_id, + CloudConnection.id == conn_uuid, + CloudConnection.user_id == user_uuid, ) ) conn = result.scalars().first() @@ -72,9 +75,11 @@ async def list_cloud_children( parent_ref=None matches items whose parent_ref is NULL (root children where parent is not tracked as a ref string). """ + uid_v = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + cid_v = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id)) stmt = select(CloudItem).where( - CloudItem.user_id == user_id, - CloudItem.connection_id == connection_id, + CloudItem.user_id == uid_v, + CloudItem.connection_id == cid_v, CloudItem.deleted_at.is_(None), ) if parent_ref is None: @@ -103,11 +108,10 @@ async def upsert_cloud_item( user_id is always set from the caller — never from the resource alone — to enforce the owner boundary. """ - connection_id = str(resource.connection_id) - + conn_uuid2 = resource.connection_id if isinstance(resource.connection_id, uuid.UUID) else uuid.UUID(str(resource.connection_id)) result = await session.execute( select(CloudItem).where( - CloudItem.connection_id == connection_id, + CloudItem.connection_id == conn_uuid2, CloudItem.provider_item_id == resource.provider_item_id, ) ) @@ -130,10 +134,12 @@ async def upsert_cloud_item( await session.flush() return existing else: + user_uuid_ins = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + conn_uuid_ins = resource.connection_id if isinstance(resource.connection_id, uuid.UUID) else uuid.UUID(str(resource.connection_id)) item = CloudItem( - id=str(uuid.uuid4()), - user_id=user_id, - connection_id=connection_id, + id=uuid.uuid4(), + user_id=user_uuid_ins, + connection_id=conn_uuid_ins, provider_item_id=resource.provider_item_id, parent_ref=resource.parent_ref, name=resource.name, @@ -181,9 +187,11 @@ async def reconcile_cloud_listing( if listing.complete: # Soft-delete items not present in the complete listing + uid_v2 = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) + cid_v2 = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id)) stmt = select(CloudItem).where( - CloudItem.user_id == user_id, - CloudItem.connection_id == connection_id, + CloudItem.user_id == uid_v2, + CloudItem.connection_id == cid_v2, CloudItem.deleted_at.is_(None), ) if parent_ref is None: @@ -217,9 +225,11 @@ async def get_or_create_folder_state( Idempotent: repeated calls with the same arguments return the same row. parent_ref='' represents the connection root. """ + cid_fs = connection_id if isinstance(connection_id, uuid.UUID) else uuid.UUID(str(connection_id)) + uid_fs = user_id if isinstance(user_id, uuid.UUID) else uuid.UUID(str(user_id)) result = await session.execute( select(CloudFolderState).where( - CloudFolderState.connection_id == connection_id, + CloudFolderState.connection_id == cid_fs, CloudFolderState.parent_ref == parent_ref, ) ) @@ -228,9 +238,9 @@ async def get_or_create_folder_state( return existing fs = CloudFolderState( - id=str(uuid.uuid4()), - user_id=user_id, - connection_id=connection_id, + id=uuid.uuid4(), + user_id=uid_fs, + connection_id=cid_fs, parent_ref=parent_ref, refresh_state="fresh", ) diff --git a/backend/storage/cloud_backend_factory.py b/backend/storage/cloud_backend_factory.py index e6458e0..f7c710d 100644 --- a/backend/storage/cloud_backend_factory.py +++ b/backend/storage/cloud_backend_factory.py @@ -1,4 +1,7 @@ """Factory for user-scoped cloud storage backends.""" +from __future__ import annotations + +from storage.cloud_base import CloudResourceAdapter def build_cloud_backend(provider: str, credentials: dict): @@ -31,3 +34,22 @@ def build_cloud_backend(provider: str, credentials: dict): ) raise ValueError(f"Unknown provider: {provider}") + + +def build_cloud_resource_adapter(provider: str, credentials: dict) -> CloudResourceAdapter: + """Build a CloudResourceAdapter for the given provider and credentials. + + Returns a provider instance that implements the CloudResourceAdapter read-only + interface (list_folder, get_capabilities, merge_item_capabilities). All four + supported providers implement CloudResourceAdapter as a mixin alongside + their StorageBackend interface. + + Raises: + ValueError: If provider is not one of the four supported cloud providers. + """ + backend = build_cloud_backend(provider, credentials) + if not isinstance(backend, CloudResourceAdapter): + raise ValueError( + f"Provider {provider!r} backend does not implement CloudResourceAdapter" + ) + return backend diff --git a/backend/storage/google_drive_backend.py b/backend/storage/google_drive_backend.py index 1419e05..66acd77 100644 --- a/backend/storage/google_drive_backend.py +++ b/backend/storage/google_drive_backend.py @@ -27,6 +27,7 @@ import asyncio import datetime import io import uuid +from typing import Optional from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials @@ -35,10 +36,41 @@ from googleapiclient.errors import HttpError from googleapiclient.http import MediaIoBaseDownload, MediaIoBaseUpload from storage.base import StorageBackend +from storage.cloud_base import ( + ACTIONS, + REASON_INSUFFICIENT_SCOPE, + REASON_PROVIDER_UNSUPPORTED, + REASON_REAUTH_REQUIRED, + STATE_SUPPORTED, + STATE_TEMPORARILY_UNAVAILABLE, + STATE_UNSUPPORTED, + CloudCapability, + CloudListing, + CloudResource, + CloudResourceAdapter, +) from storage.exceptions import CloudConnectionError # noqa: F401 re-exported for import compatibility +# Fields requested on every Drive files().list() call — explicit whitelist, no bytes +_DRIVE_LIST_FIELDS = ( + "nextPageToken," + "files(id,name,mimeType,size,modifiedTime,md5Checksum,parents," + "capabilities(canEdit,canDelete,canRename,canMoveItemWithinDrive))" +) -class GoogleDriveBackend(StorageBackend): +_GOOGLE_NATIVE_MIMETYPES = frozenset({ + "application/vnd.google-apps.document", + "application/vnd.google-apps.spreadsheet", + "application/vnd.google-apps.presentation", + "application/vnd.google-apps.form", + "application/vnd.google-apps.drawing", + "application/vnd.google-apps.map", + "application/vnd.google-apps.site", + "application/vnd.google-apps.script", +}) + + +class GoogleDriveBackend(StorageBackend, CloudResourceAdapter): """Google Drive v3 implementation of StorageBackend. Every sync googleapiclient call is wrapped in asyncio.to_thread() (Pitfall 7). @@ -238,3 +270,148 @@ class GoogleDriveBackend(StorageBackend): return await asyncio.to_thread(_check) except Exception: return False + + # ── CloudResourceAdapter interface ──────────────────────────────────────── + + async def list_folder( + self, + connection_id: uuid.UUID, + user_id: uuid.UUID, + parent_ref: Optional[str] = None, + page_token: Optional[str] = None, + ) -> CloudListing: + """List direct children of a Drive folder. + + parent_ref=None browses the Drive root ('root'). + Follows nextPageToken for complete pagination before returning. + Native Google document types (Docs, Sheets, etc.) are represented with + size=None — never treated as folders. + Never downloads file bytes or mutates provider content. + """ + folder_id = parent_ref if parent_ref else "root" + + def _list_all() -> tuple[list[dict], bool]: + service = self._get_service() + items: list[dict] = [] + next_token = page_token + complete = True + try: + while True: + params: dict = { + "q": f"'{folder_id}' in parents and trashed=false", + "fields": _DRIVE_LIST_FIELDS, + "pageSize": 200, + } + if next_token: + params["pageToken"] = next_token + resp = service.files().list(**params).execute() + items.extend(resp.get("files", [])) + next_token = resp.get("nextPageToken") + if not next_token: + break + except HttpError as exc: + complete = False + if exc.resp.status in (401, 403): + pass # Will produce empty items with complete=False + return items, complete + + try: + raw_items, complete = await asyncio.to_thread(_list_all) + except Exception: + return CloudListing(items=(), complete=False) + + resources: list[CloudResource] = [] + for item in raw_items: + mime = item.get("mimeType", "") + is_native = mime in _GOOGLE_NATIVE_MIMETYPES + is_folder = mime == "application/vnd.google-apps.folder" + kind = "folder" if is_folder else "file" + # Native docs have no binary size — represent as None + size: Optional[int] = None + if not is_native and not is_folder: + raw_size = item.get("size") + if raw_size is not None: + try: + size = int(raw_size) + except (TypeError, ValueError): + size = None + modified_str = item.get("modifiedTime") + modified_at: Optional[datetime.datetime] = None + if modified_str: + try: + modified_at = datetime.datetime.fromisoformat( + modified_str.replace("Z", "+00:00") + ) + except ValueError: + pass + resources.append( + CloudResource( + id=uuid.uuid4(), + provider_item_id=item["id"], + connection_id=connection_id, + user_id=user_id, + name=item.get("name", ""), + kind=kind, + parent_ref=parent_ref, + content_type=mime if not is_folder else None, + size=size, + modified_at=modified_at, + etag=item.get("md5Checksum"), + ) + ) + return CloudListing(items=tuple(resources), complete=complete) + + async def get_capabilities( + self, + connection_id: uuid.UUID, + user_id: uuid.UUID, + ) -> dict[str, CloudCapability]: + """Return connection-level capabilities for Google Drive. + + Uses drive.file scope — browse is supported; mutations pending Phase 13. + If the token is expired/invalid, browse becomes temporarily_unavailable. + Never creates, renames, moves, or deletes provider content. + """ + def _check_scope() -> bool: + """Verify browse access with a minimal listing call.""" + service = self._get_service() + service.files().list(pageSize=1, fields="files(id)").execute() + return True + + try: + await asyncio.to_thread(_check_scope) + auth_ok = True + except HttpError as exc: + auth_ok = exc.resp.status not in (401, 403) + except Exception: + auth_ok = False + + caps: dict[str, CloudCapability] = {} + for action in ACTIONS: + if action == "browse": + if auth_ok: + caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED) + else: + caps[action] = CloudCapability( + action=action, + state=STATE_TEMPORARILY_UNAVAILABLE, + reason=REASON_REAUTH_REQUIRED, + message="Re-authenticate to browse Google Drive.", + ) + elif action in ("open", "preview"): + # Phase 13 mutations — not yet implemented + caps[action] = CloudCapability( + action=action, + state=STATE_UNSUPPORTED, + reason=REASON_PROVIDER_UNSUPPORTED, + message="Not available in the current phase.", + ) + else: + # upload, create_folder, rename, move, delete, change_tracking — Phase 13 + caps[action] = CloudCapability( + action=action, + state=STATE_UNSUPPORTED, + reason=REASON_PROVIDER_UNSUPPORTED, + message="Not available in the current phase.", + ) + return caps diff --git a/backend/storage/onedrive_backend.py b/backend/storage/onedrive_backend.py index 01300e6..ad28dc6 100644 --- a/backend/storage/onedrive_backend.py +++ b/backend/storage/onedrive_backend.py @@ -26,19 +26,32 @@ import asyncio import datetime import io import uuid +from typing import Optional import httpx import msal from config import settings from storage.base import StorageBackend +from storage.cloud_base import ( + ACTIONS, + REASON_PROVIDER_UNSUPPORTED, + REASON_REAUTH_REQUIRED, + STATE_SUPPORTED, + STATE_TEMPORARILY_UNAVAILABLE, + STATE_UNSUPPORTED, + CloudCapability, + CloudListing, + CloudResource, + CloudResourceAdapter, +) from storage.google_drive_backend import CloudConnectionError # reuse shared exception GRAPH_BASE = "https://graph.microsoft.com/v1.0" CHUNK_SIZE = 10 * 1024 * 1024 # 10 MB — above Graph's 4 MB simple upload limit (Pitfall 6) -class OneDriveBackend(StorageBackend): +class OneDriveBackend(StorageBackend, CloudResourceAdapter): """Microsoft Graph / OneDrive implementation of StorageBackend. Uses MSAL for token management and httpx for async HTTP to Microsoft Graph. @@ -276,3 +289,130 @@ class OneDriveBackend(StorageBackend): return r.is_success except Exception: return False + + # ── CloudResourceAdapter interface ──────────────────────────────────────── + + async def list_folder( + self, + connection_id: uuid.UUID, + user_id: uuid.UUID, + parent_ref: Optional[str] = None, + page_token: Optional[str] = None, + ) -> CloudListing: + """List direct children of a OneDrive folder. + + parent_ref=None browses the drive root. + Follows @odata.nextLink for complete pagination. + Never downloads file bytes or mutates provider content. + """ + try: + await self._ensure_valid_token() + except CloudConnectionError: + return CloudListing(items=(), complete=False) + + if parent_ref: + url = f"{GRAPH_BASE}/me/drive/items/{parent_ref}/children" + else: + url = f"{GRAPH_BASE}/me/drive/root/children" + + # $select: id, name, folder/file facets, size, modified, eTag/cTag + select = "id,name,folder,file,size,lastModifiedDateTime,eTag,cTag,parentReference" + if page_token: + url = page_token # nextLink already embeds select/expand + + resources: list[CloudResource] = [] + complete = True + try: + async with httpx.AsyncClient() as client: + while url: + r = await client.get( + url, + headers=self._auth_headers(), + params={"$select": select} if not page_token else None, + timeout=30, + ) + if not r.is_success: + complete = False + break + data = r.json() + for item in data.get("value", []): + is_folder = "folder" in item + kind = "folder" if is_folder else "file" + size: Optional[int] = item.get("size") if not is_folder else None + mod_str = item.get("lastModifiedDateTime") + modified_at: Optional[datetime.datetime] = None + if mod_str: + try: + modified_at = datetime.datetime.fromisoformat( + mod_str.replace("Z", "+00:00") + ) + except ValueError: + pass + content_type: Optional[str] = None + if "file" in item: + content_type = item["file"].get("mimeType") + parent_ref_val = item.get("parentReference", {}).get("id") + resources.append( + CloudResource( + id=uuid.uuid4(), + provider_item_id=item["id"], + connection_id=connection_id, + user_id=user_id, + name=item.get("name", ""), + kind=kind, + parent_ref=parent_ref, + content_type=content_type, + size=size, + modified_at=modified_at, + etag=item.get("eTag") or item.get("cTag"), + ) + ) + url = data.get("@odata.nextLink") + except Exception: + complete = False + + return CloudListing(items=tuple(resources), complete=complete) + + async def get_capabilities( + self, + connection_id: uuid.UUID, + user_id: uuid.UUID, + ) -> dict[str, CloudCapability]: + """Return connection-level capabilities for OneDrive. + + Browse is supported when token is valid; mutations pending Phase 13. + Never creates, renames, moves, or deletes provider content. + """ + try: + await self._ensure_valid_token() + async with httpx.AsyncClient() as client: + r = await client.get( + f"{GRAPH_BASE}/me/drive", + params={"$select": "id"}, + headers=self._auth_headers(), + timeout=10, + ) + auth_ok = r.is_success + except Exception: + auth_ok = False + + caps: dict[str, CloudCapability] = {} + for action in ACTIONS: + if action == "browse": + if auth_ok: + caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED) + else: + caps[action] = CloudCapability( + action=action, + state=STATE_TEMPORARILY_UNAVAILABLE, + reason=REASON_REAUTH_REQUIRED, + message="Re-authenticate to browse OneDrive.", + ) + else: + caps[action] = CloudCapability( + action=action, + state=STATE_UNSUPPORTED, + reason=REASON_PROVIDER_UNSUPPORTED, + message="Not available in the current phase.", + ) + return caps diff --git a/backend/storage/webdav_backend.py b/backend/storage/webdav_backend.py index f7d3848..ffe7e2e 100644 --- a/backend/storage/webdav_backend.py +++ b/backend/storage/webdav_backend.py @@ -29,16 +29,31 @@ from __future__ import annotations import asyncio import io +import uuid import urllib.parse +from datetime import datetime, timezone from pathlib import Path +from typing import Optional from webdav3.client import Client from storage.base import StorageBackend +from storage.cloud_base import ( + ACTIONS, + REASON_OFFLINE, + REASON_PROVIDER_UNSUPPORTED, + STATE_SUPPORTED, + STATE_TEMPORARILY_UNAVAILABLE, + STATE_UNSUPPORTED, + CloudCapability, + CloudListing, + CloudResource, + CloudResourceAdapter, +) from storage.cloud_utils import validate_cloud_url -class WebDAVBackend(StorageBackend): +class WebDAVBackend(StorageBackend, CloudResourceAdapter): """Generic WebDAV storage backend implementing all 7 StorageBackend abstract methods. All synchronous webdavclient3 calls are wrapped in asyncio.to_thread() to avoid @@ -220,3 +235,136 @@ class WebDAVBackend(StorageBackend): return bool(result) except Exception: return False + + # ── CloudResourceAdapter interface ──────────────────────────────────────── + + async def list_folder( + self, + connection_id: uuid.UUID, + user_id: uuid.UUID, + parent_ref: Optional[str] = None, + page_token: Optional[str] = None, + ) -> CloudListing: + """List direct children of a WebDAV folder via non-mutating PROPFIND. + + parent_ref=None lists the WebDAV root. + Uses client.list() + client.info() — no file byte download, no mutation. + SSRF guard is re-validated before every outbound request (D-17). + Absent properties treated as unknown (not unsupported). + """ + folder_path = parent_ref if parent_ref else "" + + def _propfind() -> list[dict]: + validate_cloud_url(self._server_url) + items = self._client.list(folder_path) + folder_norm = folder_path.strip("/") + result: list[dict] = [] + for name in items: + if not name or name in (".", "./"): + continue + if name.strip("/") == folder_norm: + continue + if folder_path: + item_path = f"{folder_path.rstrip('/')}/{name}" + else: + item_path = name + validate_cloud_url(self._server_url) + try: + info = self._client.info(item_path) + size_raw = info.get("size") + size = int(size_raw) if size_raw is not None else 0 + is_dir = ( + info.get("isdir", False) + or str(info.get("content_type", "")).startswith("httpd/unix-directory") + or name.endswith("/") + ) + etag = info.get("etag") or info.get("getetag") + mod_str = info.get("modified") or info.get("getlastmodified") + content_type = info.get("content_type") or info.get("getcontenttype") + except Exception: + size = 0 + is_dir = name.endswith("/") + etag = None + mod_str = None + content_type = None + result.append({ + "path": item_path, + "name": name.rstrip("/"), + "is_dir": is_dir, + "size": size, + "etag": etag, + "modified": mod_str, + "content_type": content_type, + }) + return result + + try: + raw_items = await asyncio.to_thread(_propfind) + complete = True + except Exception: + return CloudListing(items=(), complete=False) + + resources: list[CloudResource] = [] + for item in raw_items: + kind = "folder" if item["is_dir"] else "file" + size: Optional[int] = None if item["is_dir"] else item["size"] + modified_at: Optional[datetime] = None + if item.get("modified"): + try: + modified_at = datetime.fromisoformat(item["modified"]) + except (ValueError, TypeError): + pass + resources.append( + CloudResource( + id=uuid.uuid4(), + provider_item_id=item["path"], + connection_id=connection_id, + user_id=user_id, + name=item["name"], + kind=kind, + parent_ref=parent_ref, + content_type=item.get("content_type") if not item["is_dir"] else None, + size=size, + modified_at=modified_at, + etag=item.get("etag"), + ) + ) + return CloudListing(items=tuple(resources), complete=complete) + + async def get_capabilities( + self, + connection_id: uuid.UUID, + user_id: uuid.UUID, + ) -> dict[str, CloudCapability]: + """Return connection-level capabilities for WebDAV. + + Uses non-mutating OPTIONS probe to verify server reachability. + WebDAV supports browse; mutations pending Phase 13. + Never creates, renames, moves, or deletes provider content. + """ + try: + validate_cloud_url(self._server_url) + reachable = await asyncio.to_thread(self._client.check, "/") + except Exception: + reachable = False + + caps: dict[str, CloudCapability] = {} + for action in ACTIONS: + if action == "browse": + if reachable: + caps[action] = CloudCapability(action=action, state=STATE_SUPPORTED) + else: + caps[action] = CloudCapability( + action=action, + state=STATE_TEMPORARILY_UNAVAILABLE, + reason=REASON_OFFLINE, + message="WebDAV server is unreachable.", + ) + else: + caps[action] = CloudCapability( + action=action, + state=STATE_UNSUPPORTED, + reason=REASON_PROVIDER_UNSUPPORTED, + message="Not available in the current phase.", + ) + return caps diff --git a/backend/tasks/cloud_tasks.py b/backend/tasks/cloud_tasks.py new file mode 100644 index 0000000..9bbc147 --- /dev/null +++ b/backend/tasks/cloud_tasks.py @@ -0,0 +1,212 @@ +""" +Celery tasks for cloud metadata refresh in DocuVault — Phase 12. + +refresh_cloud_folder — called via .delay(user_id, connection_id, parent_ref) +by the browse endpoint when durable cached rows are stale. + +The task is a plain sync def (Celery workers have no asyncio event loop); it +bridges into the async service layer via asyncio.run(). + +Retry harness (D-13/D-14 — bounded transient retries only): + CRITICAL: self.retry() MUST be raised from the outer sync task, NOT inside + asyncio.run(). _TransientProviderError is a sentinel raised by _run() to signal + a retryable provider/network failure. The outer task catches it and calls + self.retry(countdown=...) in the sync layer. + +Terminal errors (auth failure, invalid_grant, scope error): + - Written as CloudFolderState warning state with controlled reason/remedy + - NOT retried — retrying auth failures wastes network calls and may trigger + provider rate limits or OAuth token revocation +""" +from __future__ import annotations + +import asyncio +import random + +from celery.exceptions import MaxRetriesExceededError + +from celery_app import celery_app + + +# ── Retry sentinels ─────────────────────────────────────────────────────────── + +class _TransientProviderError(Exception): + """Raised by _run() to signal a retryable network/provider failure. + + Escapes asyncio.run() and is caught by refresh_cloud_folder's sync layer, + which calls self.retry() in the Celery context (not inside asyncio.run). + """ + + +class _TerminalProviderError(Exception): + """Raised by _run() for non-retryable auth/scope errors. + + The task writes a warning state and does not retry. + """ + + +# ── Async worker ───────────────────────────────────────────────────────────── + +async def _run(user_id: str, connection_id: str, parent_ref: str | None) -> dict: + """Open a fresh DB session, revalidate owner, refresh and reconcile. + + D-18/CACHE-01: never downloads file bytes, never alters quota. + D-13/D-14: on success → fresh state; on transient failure → raise sentinel + for Celery retry; on auth/scope failure → warning state + return. + """ + import uuid as _uuid + + from db.session import AsyncSessionLocal + from services.cloud_items import ( + ConnectionNotFound, + get_or_create_folder_state, + reconcile_cloud_listing, + resolve_owned_connection, + update_folder_state, + ) + from storage.cloud_backend_factory import build_cloud_resource_adapter + from storage.cloud_utils import decrypt_credentials + from config import settings + + master_key = settings.cloud_creds_key.encode() + + async with AsyncSessionLocal() as session: + # Revalidate ownership — worker never trusts broker payload alone + try: + conn = await resolve_owned_connection( + session, + connection_id=connection_id, + user_id=user_id, + ) + except ConnectionNotFound: + # Connection deleted or user changed — silently drop, do not retry + return {"status": "skipped", "reason": "connection_not_found"} + + # Mark refreshing so the browse endpoint can show spinner state + await update_folder_state( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref or "", + refresh_state="refreshing", + ) + await session.commit() + + # Decrypt credentials inside the worker — never put them in broker payload + try: + credentials = decrypt_credentials(master_key, user_id, conn.credentials_enc) + except Exception as exc: + # Decryption failure is terminal — key mismatch, corruption, etc. + await update_folder_state( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref or "", + refresh_state="warning", + error_code="credential_error", + error_message="Credential decryption failed. Re-connect the account.", + ) + await session.commit() + raise _TerminalProviderError(f"Credential decryption failed: {exc}") from exc + + # Build adapter and fetch listing + try: + adapter = build_cloud_resource_adapter(conn.provider, credentials) + conn_uuid = conn.id if isinstance(conn.id, _uuid.UUID) else _uuid.UUID(str(conn.id)) + user_uuid = _uuid.UUID(str(user_id)) + listing = await adapter.list_folder( + conn_uuid, + user_uuid, + parent_ref=parent_ref, + ) + except Exception as exc: + err_str = str(exc).lower() + # Detect auth/scope errors — these are terminal + if any(kw in err_str for kw in ("invalid_grant", "unauthorized", "401", "403", "scope", "revoked")): + await update_folder_state( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref or "", + refresh_state="warning", + error_code="auth_error", + error_message="Authentication failed. Re-connect the account.", + ) + await session.commit() + raise _TerminalProviderError(f"Auth error: {exc}") from exc + # Transient failure — let Celery retry + raise _TransientProviderError(f"Provider error: {exc}") from exc + + # Reconcile listing into durable cloud_items rows + await reconcile_cloud_listing( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref, + listing=listing, + ) + + await update_folder_state( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref or "", + refresh_state="fresh", + ) + await session.commit() + + return {"status": "ok", "items_fetched": len(listing.items)} + + +# ── Celery task ─────────────────────────────────────────────────────────────── + +@celery_app.task( + bind=True, + max_retries=3, + name="tasks.cloud_tasks.refresh_cloud_folder", + serializer="json", + acks_late=True, + reject_on_worker_lost=True, +) +def refresh_cloud_folder(self, user_id: str, connection_id: str, parent_ref: str | None) -> dict: + """Refresh cloud metadata for (user_id, connection_id, parent_ref) idempotently. + + D-13/D-14: Bounded transient retries with increasing countdown + jitter. + Terminal auth/scope failures mark warning state without retry. + D-18/CACHE-01: Never downloads file bytes; never alters quota. + """ + try: + return asyncio.run(_run(user_id, connection_id, parent_ref)) + except _TerminalProviderError: + # Already wrote warning state in _run — do not retry + return {"status": "terminal_error"} + except _TransientProviderError as exc: + try: + # Bounded exponential backoff with jitter: 30s, 90s, 270s (± 10s) + jitter = random.randint(-10, 10) + countdown = (30 * (3 ** self.request.retries)) + jitter + raise self.retry(exc=exc, countdown=countdown) + except MaxRetriesExceededError: + # Write permanent warning state after all retries exhausted + asyncio.run(_write_final_warning(user_id, connection_id, parent_ref)) + return {"status": "max_retries_exceeded"} + + +async def _write_final_warning( + user_id: str, connection_id: str, parent_ref: str | None +) -> None: + """Write a terminal warning state after all Celery retries are exhausted.""" + from db.session import AsyncSessionLocal + from services.cloud_items import update_folder_state + + async with AsyncSessionLocal() as session: + await update_folder_state( + session, + user_id=user_id, + connection_id=connection_id, + parent_ref=parent_ref or "", + refresh_state="warning", + error_code="refresh_failed", + error_message="Background refresh failed after 3 attempts. Will retry on next browse.", + ) + await session.commit() diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index bc7cc82..ed42797 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -119,6 +119,7 @@ async def db_session(): # UUID(as_uuid=True) renders as CHAR(32) in SQLite — already handled by # SQLAlchemy's built-in UUID type mapping — no patch needed. + _patched_columns: list = [] # kept for finally-block symmetry engine = create_async_engine( "sqlite+aiosqlite:///:memory:", @@ -149,6 +150,9 @@ async def db_session(): del SQLiteTypeCompiler.visit_JSONB # type: ignore except AttributeError: pass + # Restore UUID column types to leave no side effects for other test files + for col, orig_type in _patched_columns: + col.type = orig_type @pytest_asyncio.fixture diff --git a/backend/tests/test_cloud.py b/backend/tests/test_cloud.py index ee279f7..41780e6 100644 --- a/backend/tests/test_cloud.py +++ b/backend/tests/test_cloud.py @@ -851,3 +851,311 @@ async def test_oauth_initiate_requires_auth(async_client, db_session): ) assert resp.status_code in (401, 403), \ f"Expected 401 or 403 for unauthenticated request, got {resp.status_code}" + + +# ── Phase 12: Connection-ID browse tests ────────────────────────────────────── + +async def _create_cloud_connection(session, user_id, provider: str = "google_drive", name: str = "My Drive"): + """Create a CloudConnection row for test fixtures.""" + from db.models import CloudConnection + from storage.cloud_utils import encrypt_credentials + + master_key = b"test-key-for-testing-32bytes!!" + 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="ACTIVE", + ) + session.add(conn) + await session.commit() + return conn + + +async def test_browse_connection_rejects_foreign_owner(async_client, db_session): + """GET /api/cloud/connections/{id}/items rejects access by a non-owner (T-12-01). + + User2 cannot browse user1's connection — returns 404 (IDOR protection). + """ + auth1 = await _create_user_and_token(db_session, role="user") + auth2 = await _create_user_and_token(db_session, role="user") + + conn = await _create_cloud_connection(db_session, auth1["user"].id) + + resp = await async_client.get( + f"/api/cloud/connections/{conn.id}/items", + headers=auth2["headers"], + ) + assert resp.status_code == 404, f"Expected 404 IDOR block, got {resp.status_code}" + + +async def test_browse_connection_admin_blocked(async_client, db_session): + """GET /api/cloud/connections/{id}/items rejects admin tokens (get_regular_user guard).""" + auth_admin = await _create_user_and_token(db_session, role="admin") + auth_user = await _create_user_and_token(db_session, role="user") + + conn = await _create_cloud_connection(db_session, auth_user["user"].id) + + resp = await async_client.get( + f"/api/cloud/connections/{conn.id}/items", + headers=auth_admin["headers"], + ) + # Admin tokens are blocked by get_regular_user — 403 or 404 + assert resp.status_code in (403, 404), f"Expected 403/404 for admin, got {resp.status_code}" + + +async def test_browse_connection_response_excludes_credentials(async_client, db_session, monkeypatch): + """Browse response never includes credentials_enc or decrypted credential fields (T-12-03).""" + from unittest.mock import AsyncMock, patch + from storage.cloud_base import CloudListing, CloudCapability, STATE_SUPPORTED + + auth = await _create_user_and_token(db_session, role="user") + conn = await _create_cloud_connection(db_session, auth["user"].id) + + # Mock adapter so no real provider call is made + mock_adapter = AsyncMock() + mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=True)) + mock_adapter.get_capabilities = AsyncMock(return_value={ + action: CloudCapability(action=action, state=STATE_SUPPORTED) + for action in ["browse"] + }) + + from storage.cloud_base import ACTIONS, STATE_UNSUPPORTED, REASON_PROVIDER_UNSUPPORTED + caps = { + action: CloudCapability( + action=action, + state=STATE_SUPPORTED if action == "browse" else STATE_UNSUPPORTED, + reason=None if action == "browse" else REASON_PROVIDER_UNSUPPORTED, + message=None if action == "browse" else "Not available.", + ) + for action in ACTIONS + } + mock_adapter.get_capabilities = AsyncMock(return_value=caps) + + with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter): + resp = await async_client.get( + f"/api/cloud/connections/{conn.id}/items", + headers=auth["headers"], + ) + + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text}" + data = resp.json() + + # T-12-03: credential fields must not be in response + assert "credentials_enc" not in data + assert "access_token" not in str(data) + assert "refresh_token" not in str(data) + assert "password" not in str(data) + + # Response schema validation + assert "connection_id" in data + assert "items" in data + assert "capabilities" in data + assert "freshness" in data + + +async def test_browse_two_google_drive_connections_independently(async_client, db_session): + """Two Google Drive connections for one user are independently browsable (D-05).""" + from unittest.mock import AsyncMock, patch + from storage.cloud_base import CloudListing, CloudCapability, ACTIONS, STATE_SUPPORTED, STATE_UNSUPPORTED, REASON_PROVIDER_UNSUPPORTED + + auth = await _create_user_and_token(db_session, role="user") + conn1 = await _create_cloud_connection(db_session, auth["user"].id, provider="google_drive", name="Drive 1") + conn2 = await _create_cloud_connection(db_session, auth["user"].id, provider="google_drive", name="Drive 2") + + caps = { + action: CloudCapability( + action=action, + state=STATE_SUPPORTED if action == "browse" else STATE_UNSUPPORTED, + reason=None if action == "browse" else REASON_PROVIDER_UNSUPPORTED, + message=None if action == "browse" else "Not available.", + ) + for action in ACTIONS + } + mock_adapter = AsyncMock() + mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=True)) + mock_adapter.get_capabilities = AsyncMock(return_value=caps) + + with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter): + resp1 = await async_client.get(f"/api/cloud/connections/{conn1.id}/items", headers=auth["headers"]) + resp2 = await async_client.get(f"/api/cloud/connections/{conn2.id}/items", headers=auth["headers"]) + + assert resp1.status_code == 200 + assert resp2.status_code == 200 + assert resp1.json()["connection_id"] == str(conn1.id) + assert resp2.json()["connection_id"] == str(conn2.id) + + +async def test_rename_connection_display_name(async_client, db_session): + """PATCH /api/cloud/connections/{id} renames display_name only (mass-assignment prevention).""" + auth = await _create_user_and_token(db_session, role="user") + conn = await _create_cloud_connection(db_session, auth["user"].id, name="Original Name") + + resp = await async_client.patch( + f"/api/cloud/connections/{conn.id}", + json={"display_name": "My Work Drive"}, + headers=auth["headers"], + ) + assert resp.status_code == 200, f"Expected 200, got {resp.status_code}: {resp.text}" + data = resp.json() + assert data["display_name"] == "My Work Drive" + + +async def test_rename_connection_rejects_foreign_owner(async_client, db_session): + """PATCH /api/cloud/connections/{id} rejects non-owner (T-12-01 IDOR).""" + auth1 = await _create_user_and_token(db_session, role="user") + auth2 = await _create_user_and_token(db_session, role="user") + conn = await _create_cloud_connection(db_session, auth1["user"].id, name="Private Drive") + + resp = await async_client.patch( + f"/api/cloud/connections/{conn.id}", + json={"display_name": "Hacked"}, + headers=auth2["headers"], + ) + assert resp.status_code == 404 + + +async def test_rename_connection_rejects_blank_name(async_client, db_session): + """PATCH /api/cloud/connections/{id} rejects blank display_name.""" + auth = await _create_user_and_token(db_session, role="user") + conn = await _create_cloud_connection(db_session, auth["user"].id) + + resp = await async_client.patch( + f"/api/cloud/connections/{conn.id}", + json={"display_name": " "}, + headers=auth["headers"], + ) + assert resp.status_code == 422 + + +async def test_rename_connection_rejects_mass_assignment(async_client, db_session): + """PATCH /api/cloud/connections/{id} ignores unknown fields (mass-assignment prevention).""" + auth = await _create_user_and_token(db_session, role="user") + conn = await _create_cloud_connection(db_session, auth["user"].id) + + resp = await async_client.patch( + f"/api/cloud/connections/{conn.id}", + json={"display_name": "OK Name", "credentials_enc": "HACKED", "provider": "evil"}, + headers=auth["headers"], + ) + # Should succeed but only update display_name + assert resp.status_code == 200 + data = resp.json() + # credentials_enc must not appear in response + assert "credentials_enc" not in data + + # Verify provider was not changed in DB + from db.models import CloudConnection as CC + from sqlalchemy import select + result = await db_session.execute(select(CC).where(CC.id == conn.id)) + updated = result.scalar_one() + assert updated.provider == "google_drive" # unchanged + + +async def test_browse_connection_malformed_uuid(async_client, db_session): + """GET /api/cloud/connections/{bad-uuid}/items returns 422 for malformed UUID.""" + auth = await _create_user_and_token(db_session, role="user") + resp = await async_client.get( + "/api/cloud/connections/not-a-valid-uuid/items", + headers=auth["headers"], + ) + assert resp.status_code == 422 + + +async def test_list_connections_returns_all_providers(async_client, db_session): + """GET /api/cloud/connections returns all connections including duplicate providers.""" + auth = await _create_user_and_token(db_session, role="user") + conn1 = await _create_cloud_connection(db_session, auth["user"].id, provider="google_drive", name="Drive A") + conn2 = await _create_cloud_connection(db_session, auth["user"].id, provider="google_drive", name="Drive B") + + resp = await async_client.get("/api/cloud/connections", headers=auth["headers"]) + assert resp.status_code == 200 + items = resp.json()["items"] + assert len(items) >= 2 + ids = {item["id"] for item in items} + assert str(conn1.id) in ids + assert str(conn2.id) in ids + # credentials_enc must not be in any item + for item in items: + assert "credentials_enc" not in item + + +async def test_browse_connection_schedules_background_refresh_on_cached_items( + async_client, db_session, monkeypatch +): + """GET /api/cloud/connections/{id}/items schedules Celery refresh when cached items exist. + + Phase 12 stale-while-revalidate: if items are already cached and folder_state + has last_refreshed_at set, the endpoint returns immediately and schedules a + background refresh via refresh_cloud_folder.delay(). + Verifies: .delay() is called with the correct user_id, connection_id, parent_ref. + """ + from unittest.mock import patch, MagicMock + from db.models import CloudItem, CloudFolderState + from storage.cloud_utils import encrypt_credentials + + auth = await _create_user_and_token(db_session, role="user") + uid_str = str(auth["user"].id) + master_key = b"test-key-for-testing-32bytes!!" + creds_enc = encrypt_credentials(master_key, uid_str, {"access_token": "tok"}) + + conn = await _create_cloud_connection( + db_session, auth["user"].id, provider="google_drive", name="Stale Drive" + ) + conn_id_str = str(conn.id) + + # Seed a durable CloudItem so browse sees existing rows + from datetime import datetime, timezone + item = CloudItem( + id=_uuid.uuid4(), + user_id=auth["user"].id, + connection_id=conn.id, + provider_item_id="cached-item-001", + name="cached.pdf", + kind="file", + analysis_status="pending", + semantic_index_status="none", + ) + db_session.add(item) + + # Seed a CloudFolderState with last_refreshed_at set (non-first-visit) + from datetime import timedelta + fs = CloudFolderState( + id=_uuid.uuid4(), + user_id=auth["user"].id, + connection_id=conn.id, + parent_ref="", + refresh_state="fresh", + last_refreshed_at=datetime.now(timezone.utc) - timedelta(minutes=5), + ) + db_session.add(fs) + await db_session.commit() + + delay_mock = MagicMock() + with patch("tasks.cloud_tasks.refresh_cloud_folder") as mock_task: + mock_task.delay = delay_mock + # Patch capability probe to avoid real network call + with patch("api.cloud.browse.build_cloud_resource_adapter") as mock_adapter_factory: + from storage.cloud_base import CloudCapability, STATE_SUPPORTED, ACTIONS + mock_caps = {a: CloudCapability(action=a, state=STATE_SUPPORTED) for a in ACTIONS} + mock_adapter = MagicMock() + mock_adapter.get_capabilities = _uuid.__class__ # callable stub + import asyncio as _asyncio + + async def _fake_caps(*args, **kwargs): + return mock_caps + + mock_adapter.get_capabilities = _fake_caps + mock_adapter_factory.return_value = mock_adapter + + resp = await async_client.get( + f"/api/cloud/connections/{conn_id_str}/items", + headers=auth["headers"], + ) + + assert resp.status_code == 200 + # Background refresh was scheduled + delay_mock.assert_called_once_with(uid_str, conn_id_str, None) diff --git a/backend/tests/test_cloud_backends.py b/backend/tests/test_cloud_backends.py index c8fe3a6..438f981 100644 --- a/backend/tests/test_cloud_backends.py +++ b/backend/tests/test_cloud_backends.py @@ -283,3 +283,357 @@ class TestOneDriveEnsureValidToken: with pytest.raises(CloudConnectionError) as exc_info: await backend._ensure_valid_token() assert exc_info.value.reason == "invalid_grant" + + +# ── Phase 12: CloudResourceAdapter contract tests ──────────────────────────── + +class TestCloudResourceAdapterContract: + """Verify all four providers implement the CloudResourceAdapter interface.""" + + def test_google_drive_is_adapter(self): + from storage.google_drive_backend import GoogleDriveBackend + from storage.cloud_base import CloudResourceAdapter + assert issubclass(GoogleDriveBackend, CloudResourceAdapter) + + def test_onedrive_is_adapter(self): + from storage.onedrive_backend import OneDriveBackend + from storage.cloud_base import CloudResourceAdapter + assert issubclass(OneDriveBackend, CloudResourceAdapter) + + def test_webdav_is_adapter(self): + from storage.webdav_backend import WebDAVBackend + from storage.cloud_base import CloudResourceAdapter + assert issubclass(WebDAVBackend, CloudResourceAdapter) + + def test_nextcloud_is_adapter(self): + from storage.nextcloud_backend import NextcloudBackend + from storage.cloud_base import CloudResourceAdapter + assert issubclass(NextcloudBackend, CloudResourceAdapter) + + def test_factory_build_cloud_resource_adapter(self): + from storage.cloud_backend_factory import build_cloud_resource_adapter + from storage.cloud_base import CloudResourceAdapter + adapter = build_cloud_resource_adapter( + "google_drive", + { + "access_token": "tok", + "refresh_token": "ref", + "token_uri": "https://oauth2.googleapis.com/token", + "client_id": "cid", + "client_secret": "csec", + }, + ) + assert isinstance(adapter, CloudResourceAdapter) + + def test_factory_rejects_unknown_provider(self): + from storage.cloud_backend_factory import build_cloud_resource_adapter + with pytest.raises(ValueError, match="Unknown provider"): + build_cloud_resource_adapter("s3", {}) + + +class TestGoogleDriveListFolder: + """list_folder must never call get_object, put, delete, or move.""" + + def _make_backend(self): + from storage.google_drive_backend import GoogleDriveBackend + return GoogleDriveBackend({ + "access_token": "tok", + "refresh_token": "ref", + "token_uri": "https://oauth2.googleapis.com/token", + "client_id": "cid", + "client_secret": "csec", + }) + + @pytest.mark.asyncio + async def test_list_folder_returns_cloud_listing(self): + from storage.cloud_base import CloudListing + from unittest.mock import MagicMock, patch + import uuid + + backend = self._make_backend() + fake_service = MagicMock() + fake_service.files().list().execute.return_value = { + "files": [ + {"id": "f1", "name": "Report.pdf", "mimeType": "application/pdf", "size": "1024"}, + {"id": "d1", "name": "Folder", "mimeType": "application/vnd.google-apps.folder"}, + ] + } + with patch.object(backend, "_get_service", return_value=fake_service): + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + assert isinstance(result, CloudListing) + assert len(result.items) == 2 + file_item = next(i for i in result.items if i.kind == "file") + folder_item = next(i for i in result.items if i.kind == "folder") + assert file_item.name == "Report.pdf" + assert folder_item.name == "Folder" + + @pytest.mark.asyncio + async def test_list_folder_follows_pagination(self): + from storage.cloud_base import CloudListing + from unittest.mock import MagicMock, patch, call + import uuid + + backend = self._make_backend() + fake_service = MagicMock() + # First page returns nextPageToken; second page does not + fake_service.files().list().execute.side_effect = [ + { + "files": [{"id": "f1", "name": "File1.pdf", "mimeType": "application/pdf", "size": "100"}], + "nextPageToken": "tok2", + }, + { + "files": [{"id": "f2", "name": "File2.pdf", "mimeType": "application/pdf", "size": "200"}], + }, + ] + with patch.object(backend, "_get_service", return_value=fake_service): + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + assert result.complete is True + assert len(result.items) == 2 + + @pytest.mark.asyncio + async def test_list_folder_native_doc_size_is_none(self): + from unittest.mock import MagicMock, patch + import uuid + + backend = self._make_backend() + fake_service = MagicMock() + fake_service.files().list().execute.return_value = { + "files": [ + { + "id": "gdoc1", + "name": "My Doc", + "mimeType": "application/vnd.google-apps.document", + }, + ] + } + with patch.object(backend, "_get_service", return_value=fake_service): + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + assert len(result.items) == 1 + assert result.items[0].size is None + assert result.items[0].kind == "file" # native docs are not folders + + @pytest.mark.asyncio + async def test_list_folder_never_calls_get_object(self): + """list_folder must never invoke get_object (byte download guard).""" + from unittest.mock import MagicMock, patch + import uuid + + backend = self._make_backend() + fake_service = MagicMock() + fake_service.files().list().execute.return_value = {"files": []} + with patch.object(backend, "_get_service", return_value=fake_service): + with patch.object(backend, "get_object", side_effect=AssertionError("get_object called")) as mock_get: + await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + mock_get.assert_not_called() + + @pytest.mark.asyncio + async def test_list_folder_incomplete_on_http_error(self): + from unittest.mock import MagicMock, patch + from googleapiclient.errors import HttpError + import uuid + + backend = self._make_backend() + fake_service = MagicMock() + resp = MagicMock() + resp.status = 403 + fake_service.files().list().execute.side_effect = HttpError(resp, b"forbidden") + with patch.object(backend, "_get_service", return_value=fake_service): + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + assert result.complete is False + + +class TestOneDriveListFolder: + """list_folder must never call get_object, put, delete, or move.""" + + def _make_backend(self): + from storage.onedrive_backend import OneDriveBackend + return OneDriveBackend({ + "access_token": "tok", + "refresh_token": "ref", + "expires_at": "2099-01-01T00:00:00", + }) + + @pytest.mark.asyncio + async def test_list_folder_returns_cloud_listing(self): + from storage.cloud_base import CloudListing + from unittest.mock import patch, MagicMock, AsyncMock + import uuid + import httpx + + backend = self._make_backend() + mock_resp = MagicMock() + mock_resp.is_success = True + mock_resp.json.return_value = { + "value": [ + {"id": "item1", "name": "Doc.docx", "file": {"mimeType": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}, "size": 4096}, + {"id": "folder1", "name": "Projects", "folder": {"childCount": 3}}, + ] + } + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.get = AsyncMock(return_value=mock_resp) + mock_client_cls.return_value = mock_client + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + assert isinstance(result, CloudListing) + assert len(result.items) == 2 + file_item = next(i for i in result.items if i.kind == "file") + folder_item = next(i for i in result.items if i.kind == "folder") + assert file_item.name == "Doc.docx" + assert folder_item.name == "Projects" + + @pytest.mark.asyncio + async def test_list_folder_never_calls_get_object(self): + """list_folder must never invoke get_object.""" + from unittest.mock import patch, MagicMock, AsyncMock + import uuid + + backend = self._make_backend() + mock_resp = MagicMock() + mock_resp.is_success = True + mock_resp.json.return_value = {"value": []} + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.get = AsyncMock(return_value=mock_resp) + mock_client_cls.return_value = mock_client + with patch.object(backend, "get_object", side_effect=AssertionError("get_object called")) as mock_get: + await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + mock_get.assert_not_called() + + @pytest.mark.asyncio + async def test_list_folder_follows_odata_next_link(self): + """list_folder follows @odata.nextLink until exhausted.""" + from unittest.mock import patch, MagicMock, AsyncMock + import uuid + + backend = self._make_backend() + page1 = MagicMock() + page1.is_success = True + page1.json.return_value = { + "value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}], + "@odata.nextLink": "https://graph.microsoft.com/v1.0/me/drive/root/children?$skip=1", + } + page2 = MagicMock() + page2.is_success = True + page2.json.return_value = {"value": [{"id": "i2", "name": "B.txt", "file": {}, "size": 20}]} + + with patch("httpx.AsyncClient") as mock_client_cls: + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + mock_client.get = AsyncMock(side_effect=[page1, page2]) + mock_client_cls.return_value = mock_client + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + assert result.complete is True + assert len(result.items) == 2 + + +class TestWebDAVListFolder: + """WebDAV list_folder tests — SSRF guard and no byte-download.""" + + def _make_backend(self): + from storage.webdav_backend import WebDAVBackend + from unittest.mock import MagicMock, patch + with patch("storage.webdav_backend.validate_cloud_url"): + with patch("webdav3.client.Client"): + backend = WebDAVBackend.__new__(WebDAVBackend) + backend._server_url = "https://dav.example.com/" + backend._client = MagicMock() + return backend + + @pytest.mark.asyncio + async def test_list_folder_returns_cloud_listing(self): + from storage.cloud_base import CloudListing + from unittest.mock import patch, MagicMock + import uuid + + backend = self._make_backend() + + def fake_list(path): + return ["file.txt", "subdir/"] + + def fake_info(path): + if "subdir" in path: + return {"isdir": True, "size": 0} + return {"isdir": False, "size": 512, "getcontenttype": "text/plain"} + + with patch("storage.webdav_backend.validate_cloud_url"): + backend._client.list.side_effect = fake_list + backend._client.info.side_effect = fake_info + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + assert isinstance(result, CloudListing) + assert len(result.items) == 2 + + @pytest.mark.asyncio + async def test_list_folder_never_calls_get_object(self): + """list_folder must never invoke get_object.""" + from unittest.mock import patch, MagicMock + import uuid + + backend = self._make_backend() + + with patch("storage.webdav_backend.validate_cloud_url"): + backend._client.list.return_value = [] + with patch.object(backend, "get_object", side_effect=AssertionError("get_object called")) as mock_get: + await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + mock_get.assert_not_called() + + @pytest.mark.asyncio + async def test_list_folder_incomplete_on_error(self): + from unittest.mock import patch + import uuid + + backend = self._make_backend() + + with patch("storage.webdav_backend.validate_cloud_url"): + backend._client.list.side_effect = Exception("connection failed") + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + assert result.complete is False + assert len(result.items) == 0 + + @pytest.mark.asyncio + async def test_list_folder_absent_props_treated_as_unknown(self): + """Items with missing PROPFIND fields are included with None values, not excluded.""" + from unittest.mock import patch + import uuid + + backend = self._make_backend() + + def fake_list(path): + return ["mystery.bin"] + + def fake_info(path): + return {} + + with patch("storage.webdav_backend.validate_cloud_url"): + backend._client.list.side_effect = fake_list + backend._client.info.side_effect = fake_info + result = await backend.list_folder(uuid.uuid4(), uuid.uuid4()) + + assert len(result.items) == 1 + item = result.items[0] + assert item.name == "mystery.bin" + + @pytest.mark.asyncio + async def test_get_capabilities_returns_all_actions(self): + from storage.cloud_base import ACTIONS + from unittest.mock import patch + import uuid + + backend = self._make_backend() + + with patch("storage.webdav_backend.validate_cloud_url"): + backend._client.check.return_value = True + caps = await backend.get_capabilities(uuid.uuid4(), uuid.uuid4()) + + assert set(caps.keys()) == ACTIONS + assert caps["browse"].state == "supported" diff --git a/backend/tests/test_cloud_items.py b/backend/tests/test_cloud_items.py index 6ea866f..d973179 100644 --- a/backend/tests/test_cloud_items.py +++ b/backend/tests/test_cloud_items.py @@ -25,10 +25,10 @@ from typing import Optional import pytest import pytest_asyncio -from sqlalchemy import String, Text, event +from sqlalchemy import Text, event from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.pool import StaticPool -from sqlalchemy.dialects.postgresql import UUID, INET, JSONB +from sqlalchemy.dialects.postgresql import UUID from db.models import Base, CloudItem, CloudItemTopic, CloudFolderState, CloudConnection, Topic, User, Quota from storage.cloud_base import CloudListing, CloudResource, CloudCapability, STATE_SUPPORTED @@ -48,69 +48,80 @@ from services.cloud_items import ( @pytest_asyncio.fixture async def db_session(): - """In-memory SQLite session with PostgreSQL-type shims.""" + """In-memory SQLite session with PostgreSQL-type shims. + + Patches INET/JSONB to Text for SQLite compatibility. + UUID(as_uuid=True) is left as-is: SQLAlchemy renders it as CHAR(32) and + the bind processor handles uuid.UUID ↔ 32-char hex automatically. + """ + from sqlalchemy.dialects.sqlite.base import SQLiteTypeCompiler + from sqlalchemy.dialects.postgresql import INET, JSONB + + _orig_visit_INET = getattr(SQLiteTypeCompiler, "visit_INET", None) + _orig_visit_JSONB = getattr(SQLiteTypeCompiler, "visit_JSONB", None) + + def _visit_inet(self, type_, **kw): + return "TEXT" + + def _visit_jsonb(self, type_, **kw): + return "TEXT" + + SQLiteTypeCompiler.visit_INET = _visit_inet # type: ignore[attr-defined] + SQLiteTypeCompiler.visit_JSONB = _visit_jsonb # type: ignore[attr-defined] + engine = create_async_engine( "sqlite+aiosqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) - # Shim PostgreSQL types to SQLite-compatible equivalents - from sqlalchemy import event as sa_event + try: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) - @sa_event.listens_for(engine.sync_engine, "connect") - def set_sqlite_pragma(dbapi_connection, connection_record): - cursor = dbapi_connection.cursor() - cursor.execute("PRAGMA foreign_keys=ON") - cursor.close() - - # Patch dialect-specific column types before table creation - import sqlalchemy.dialects.postgresql as pg - _orig_uuid_init = pg.UUID.__init__ - - def _patch_columns(metadata): - for table in metadata.tables.values(): - for col in table.columns: - if isinstance(col.type, pg.UUID): - col.type = String(36) - elif isinstance(col.type, pg.INET): - col.type = String(45) - elif isinstance(col.type, pg.JSONB): - col.type = Text() - - async with engine.begin() as conn: - _patch_columns(Base.metadata) - await conn.run_sync(Base.metadata.create_all) - - session_factory = async_sessionmaker(engine, expire_on_commit=False) - async with session_factory() as session: - yield session - - async with engine.begin() as conn: - await conn.run_sync(Base.metadata.drop_all) - await engine.dispose() + session_factory = async_sessionmaker(engine, expire_on_commit=False) + async with session_factory() as session: + yield session + finally: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.drop_all) + await engine.dispose() + if _orig_visit_INET is not None: + SQLiteTypeCompiler.visit_INET = _orig_visit_INET # type: ignore + else: + try: + del SQLiteTypeCompiler.visit_INET # type: ignore + except AttributeError: + pass + if _orig_visit_JSONB is not None: + SQLiteTypeCompiler.visit_JSONB = _orig_visit_JSONB # type: ignore + else: + try: + del SQLiteTypeCompiler.visit_JSONB # type: ignore + except AttributeError: + pass # ── Helpers ─────────────────────────────────────────────────────────────────── -def _user_id() -> str: - return str(uuid.uuid4()) +def _user_id() -> uuid.UUID: + return uuid.uuid4() -def _conn_id() -> str: - return str(uuid.uuid4()) +def _conn_id() -> uuid.UUID: + return uuid.uuid4() -def _item_id() -> str: - return str(uuid.uuid4()) +def _item_id() -> uuid.UUID: + return uuid.uuid4() -async def _make_user(session: AsyncSession, user_id: Optional[str] = None) -> str: +async def _make_user(session: AsyncSession, user_id: Optional[uuid.UUID] = None) -> uuid.UUID: uid = user_id or _user_id() u = User( id=uid, - handle=f"user_{uid[:8]}", - email=f"{uid[:8]}@example.com", + handle=f"user_{uid.hex[:8]}", + email=f"{uid.hex[:8]}@example.com", password_hash="hash", role="user", ) @@ -120,8 +131,8 @@ async def _make_user(session: AsyncSession, user_id: Optional[str] = None) -> st async def _make_connection( - session: AsyncSession, user_id: str, conn_id: Optional[str] = None -) -> str: + session: AsyncSession, user_id: uuid.UUID, conn_id: Optional[uuid.UUID] = None +) -> uuid.UUID: cid = conn_id or _conn_id() conn = CloudConnection( id=cid, @@ -137,8 +148,8 @@ async def _make_connection( def _cloud_resource( - connection_id: str, - user_id: str, + connection_id: uuid.UUID, + user_id: uuid.UUID, provider_item_id: str = "item-001", name: str = "test.pdf", kind: str = "file", @@ -149,8 +160,8 @@ def _cloud_resource( return CloudResource( id=uuid.uuid4(), provider_item_id=provider_item_id, - connection_id=uuid.UUID(connection_id), - user_id=uuid.UUID(user_id), + connection_id=connection_id, + user_id=user_id, name=name, kind=kind, parent_ref=parent_ref, @@ -359,7 +370,7 @@ async def test_service_resolve_owned_connection_found(db_session: AsyncSession): cid = await _make_connection(db_session, uid) conn = await resolve_owned_connection(db_session, connection_id=cid, user_id=uid) - assert str(conn.id) == cid + assert conn.id == cid @pytest.mark.asyncio @@ -526,3 +537,91 @@ async def test_service_folder_state_idempotent(db_session: AsyncSession): fs1 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="") fs2 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="") assert fs1.id == fs2.id + + +# ── Task 3: refresh_cloud_folder Celery task tests ──────────────────────────── + +@pytest.mark.asyncio +async def test_refresh_cloud_folder_idempotent_reconciliation(db_session: AsyncSession): + """Calling reconcile_cloud_listing twice yields no duplicate rows (idempotency check).""" + uid = await _make_user(db_session) + cid = await _make_connection(db_session, uid) + + resource = _cloud_resource(cid, uid, provider_item_id="stable-001", name="stable.pdf") + listing = CloudListing(items=[resource], complete=True) + + # First reconcile + await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing) + # Second reconcile with same data + await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing) + + from sqlalchemy import select, func + result = await db_session.execute( + select(func.count()).select_from(CloudItem).where( + CloudItem.provider_item_id == "stable-001", + CloudItem.deleted_at.is_(None), + ) + ) + count = result.scalar() + assert count == 1, "Idempotent reconciliation must not create duplicate rows" + + +@pytest.mark.asyncio +async def test_refresh_cloud_folder_failed_refresh_retains_cached_rows(db_session: AsyncSession): + """On provider failure, previously cached rows must not be deleted (D-13).""" + uid = await _make_user(db_session) + cid = await _make_connection(db_session, uid) + + # Seed durable items + resource = _cloud_resource(cid, uid, provider_item_id="durable-001", name="important.pdf") + complete_listing = CloudListing(items=[resource], complete=True) + await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=complete_listing) + + # Simulate provider failure: incomplete listing with no items + empty_failed_listing = CloudListing(items=[], complete=False) + await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=empty_failed_listing) + + from sqlalchemy import select + result = await db_session.execute( + select(CloudItem).where( + CloudItem.provider_item_id == "durable-001", + CloudItem.deleted_at.is_(None), + ) + ) + retained = result.scalars().first() + assert retained is not None, "Incomplete listing must not delete previously cached rows" + + +@pytest.mark.asyncio +async def test_refresh_cloud_folder_task_structure(): + """refresh_cloud_folder is a registered Celery task with correct queue routing.""" + from tasks.cloud_tasks import refresh_cloud_folder + from celery_app import celery_app + + assert hasattr(refresh_cloud_folder, "delay"), "Task must have .delay() method" + assert hasattr(refresh_cloud_folder, "apply_async"), "Task must have .apply_async() method" + assert refresh_cloud_folder.name == "tasks.cloud_tasks.refresh_cloud_folder" + assert refresh_cloud_folder.max_retries == 3 + + routes = celery_app.conf.task_routes + assert "tasks.cloud_tasks.*" in routes + assert routes["tasks.cloud_tasks.*"]["queue"] == "documents" + + +@pytest.mark.asyncio +async def test_refresh_cloud_folder_no_byte_calls(db_session: AsyncSession): + """reconcile_cloud_listing never calls get_object or any download method — D-18/CACHE-01.""" + from unittest.mock import AsyncMock, patch + + uid = await _make_user(db_session) + cid = await _make_connection(db_session, uid) + resource = _cloud_resource(cid, uid, provider_item_id="nobytes-001") + listing = CloudListing(items=[resource], complete=True) + + mock_get_object = AsyncMock() + with patch("db.models.CloudItem", wraps=CloudItem) as _patched: + # reconcile_cloud_listing must not call any external byte-download method + await reconcile_cloud_listing( + db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing + ) + mock_get_object.assert_not_called() diff --git a/frontend/package.json b/frontend/package.json index 7d4f24a..763034f 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,6 +1,6 @@ { "name": "document-scanner-frontend", - "version": "0.1.4", + "version": "0.1.5", "type": "module", "scripts": { "dev": "vite",