chore: merge executor worktree (worktree-agent-a032d981c3b3ec11b)

This commit is contained in:
curo1305
2026-06-18 22:39:06 +02:00
7 changed files with 1852 additions and 0 deletions
@@ -0,0 +1,121 @@
---
phase: "12"
plan: "01"
subsystem: cloud-resource-foundation
tags: [cloud, metadata, capability, schema, migration, service]
dependency_graph:
requires: []
provides:
- backend/storage/cloud_base.py — CloudCapability, CloudResource, CloudListing, CloudResourceAdapter
- backend/migrations/versions/0006_cloud_resource_foundation.py — cloud_items, cloud_item_topics, cloud_folder_states tables
- backend/db/models.py — CloudItem, CloudItemTopic, CloudFolderState ORM models
- backend/services/cloud_items.py — reconcile_cloud_listing, upsert_cloud_item, resolve_owned_connection
affects:
- backend/db/models.py — extended with 3 new ORM classes
tech_stack:
added:
- SQLAlchemy 2.0 async ORM models for cloud metadata
- Alembic migration 0006 with cloud_items, cloud_item_topics, cloud_folder_states
patterns:
- Frozen dataclass value types for immutable capability/resource contracts
- Composite (connection_id, provider_item_id) uniqueness for stable item identity
- Soft-deletion reconciliation gated on CloudListing.complete=True
- Domain exceptions (ConnectionNotFound, CloudItemNotFound) — no HTTPException in service
key_files:
created:
- backend/storage/cloud_base.py
- backend/migrations/versions/0006_cloud_resource_foundation.py
- backend/services/cloud_items.py
- backend/tests/test_cloud_capabilities.py
- backend/tests/test_cloud_items.py
modified:
- backend/db/models.py
decisions:
- Frozen dataclasses for CloudCapability/CloudResource/CloudListing enforce immutability at Python level
- parent_ref='' (empty string) represents connection root in cloud_folder_states to allow unique constraint
- provider_size never referenced in quota service — D-18 compliance by design
- No object_key field on CloudItem — provider bytes are not mirrored in Phase 12
- display_name_override column added to cloud_connections for user-customized names without uniqueness changes
metrics:
duration: "~15 minutes"
completed: "2026-06-18"
tasks_completed: 3
tasks_total: 3
files_created: 5
files_modified: 1
tests_added: 46
---
# Phase 12 Plan 01: Cloud Resource Foundation Summary
**One-liner:** Provider-neutral CloudResourceAdapter contract, Alembic migration 0006 with durable owner-scoped cloud_items/cloud_folder_states tables, and idempotent reconciliation service with stable UUID identity across provider rename/move.
## What Was Built
### Task 1: Normalized cloud resource capabilities (0a7273b)
`backend/storage/cloud_base.py` defines the Phase 12 read-only contract:
- 9 action keys (browse, open, preview, upload, create_folder, rename, move, delete, change_tracking)
- 3 capability states (supported, unsupported, temporarily_unavailable)
- 6 reason codes (provider_unsupported, insufficient_scope, read_only, reauth_required, offline, item_restricted)
- Frozen dataclasses: `CloudCapability`, `CloudResource`, `CloudListing`
- Abstract `CloudResourceAdapter` with `list_folder`, `get_capabilities`, `merge_item_capabilities`
- No mutation methods in Phase 12 interface — Phase 13 boundary enforced
29 unit tests covering vocabulary, validation, merge behavior, and a fake adapter proving no mutation methods exist.
### Task 2: Durable owner-scoped metadata schema (718fb2c)
Migration `0006_cloud_resource_foundation.py` adds:
- `cloud_items`: UUID PK, user_id + connection_id ownership FKs with CASCADE, unique (connection_id, provider_item_id), no object_key or retained-byte field
- `cloud_item_topics`: association between CloudItem and Topic without requiring a Document row
- `cloud_folder_states`: per-connection/parent-ref freshness row; parent_ref='' for root enables unique constraint
- `cloud_connections`: `display_name_override` column for user-defined same-provider disambiguation
ORM models `CloudItem`, `CloudItemTopic`, `CloudFolderState` added to `backend/db/models.py`.
### Task 3: Owner-scoped reconciliation service (718fb2c)
`backend/services/cloud_items.py` implements:
- `resolve_owned_connection(session, connection_id, user_id)` — raises `ConnectionNotFound` for cross-owner access
- `list_cloud_children(session, user_id, connection_id, parent_ref)` — composite owner+connection scope
- `upsert_cloud_item(session, user_id, resource)` — preserves CloudItem UUID across rename/move
- `reconcile_cloud_listing(session, ...)` — soft-deletes missing items only when `CloudListing.complete=True`
- `get_or_create_folder_state` / `update_folder_state` — idempotent, controlled error_code/message
17 tests covering rename/move stable identity, complete removal, incomplete-listing retention, owner isolation, and idempotency.
## Verification
- `pytest -q tests/test_cloud_capabilities.py tests/test_cloud_items.py` — 46 passed
- `rg "HTTPException" backend/services/cloud_items.py` — no matches
- No Phase 12 task writes provider file bytes or MinIO objects
## Deviations from Plan
None — plan executed exactly as written. Service module (`cloud_items.py`) was created during Task 2 preparation since tests required the import; it was committed in the Task 2 commit which also covered the Task 3 deliverable.
## Threat Flags
None. All T-12-01 through T-12-09 mitigations addressed:
- T-12-01: Composite owner+connection predicates enforced in every service function
- T-12-02: Connection UUID in uniqueness constraint and ownership checks
- T-12-06: Read-only interface verified by fake adapter test
- T-12-07: Deletion gated on `complete=True` in `reconcile_cloud_listing`
- T-12-09: `provider_size` never touches quota service — confirmed by `test_model_quota_unchanged_after_item_upsert`
## Self-Check: PASSED
Files exist:
- backend/storage/cloud_base.py — FOUND
- backend/migrations/versions/0006_cloud_resource_foundation.py — FOUND
- backend/services/cloud_items.py — FOUND
- backend/tests/test_cloud_capabilities.py — FOUND
- backend/tests/test_cloud_items.py — FOUND
Commits:
- 0a7273b — feat(12-01): normalized cloud resource capability contract and unit tests
- 718fb2c — feat(12-01): durable owner-scoped cloud metadata schema (migration 0006 + models)
+138
View File
@@ -318,6 +318,144 @@ class CloudConnection(Base):
__table_args__ = (Index("ix_cloud_connections_user", "user_id"),)
class CloudItem(Base):
"""Durable per-item cloud metadata indexed by (connection_id, provider_item_id).
Phase 12: metadata browsing only — no MinIO object_key field.
Phase 14: extracted_text, analysis_status, semantic_index_status, semantic_index_data
are reserved for byte analysis and semantic search.
D-18: provider_size never flows to quotas.used_bytes.
"""
__tablename__ = "cloud_items"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
)
connection_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("cloud_connections.id", ondelete="CASCADE"),
nullable=False,
)
provider_item_id: Mapped[str] = mapped_column(Text, nullable=False)
parent_ref: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
path_snapshot: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
name: Mapped[str] = mapped_column(Text, nullable=False)
kind: Mapped[str] = mapped_column(String(8), nullable=False) # "file" | "folder"
content_type: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
# Provider-reported size — never used to update quotas.used_bytes (D-18)
provider_size: Mapped[Optional[int]] = mapped_column(BigInteger, nullable=True)
etag: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
version: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
modified_at: Mapped[Optional[datetime]] = mapped_column(
TIMESTAMP(timezone=True), nullable=True
)
last_seen_at: Mapped[datetime] = mapped_column(
TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
)
deleted_at: Mapped[Optional[datetime]] = mapped_column(
TIMESTAMP(timezone=True), nullable=True
)
# Phase 14 reserved fields
extracted_text: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
analysis_status: Mapped[str] = mapped_column(
String(16), nullable=False, default="pending"
)
semantic_index_status: Mapped[str] = mapped_column(
String(16), nullable=False, default="none"
)
semantic_index_data: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True)
created_at: Mapped[datetime] = mapped_column(
TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
)
__table_args__ = (
UniqueConstraint(
"connection_id",
"provider_item_id",
name="uq_cloud_items_connection_provider_item",
),
Index("ix_cloud_items_user_id", "user_id"),
Index("ix_cloud_items_connection_id", "connection_id"),
Index("ix_cloud_items_connection_parent", "connection_id", "parent_ref"),
)
class CloudItemTopic(Base):
"""Association between CloudItem and Topic — no Document row required."""
__tablename__ = "cloud_item_topics"
cloud_item_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("cloud_items.id", ondelete="CASCADE"),
primary_key=True,
)
topic_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("topics.id", ondelete="CASCADE"),
primary_key=True,
)
class CloudFolderState(Base):
"""Per-connection folder freshness row for cached-first navigation.
parent_ref = '' represents the connection root (allows unique constraint).
refresh_state: "refreshing" | "fresh" | "warning"
"""
__tablename__ = "cloud_folder_states"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
)
user_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
)
connection_id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
ForeignKey("cloud_connections.id", ondelete="CASCADE"),
nullable=False,
)
# '' for root, provider ref for non-root
parent_ref: Mapped[str] = mapped_column(Text, nullable=False, default="")
refresh_state: Mapped[str] = mapped_column(
String(16), nullable=False, default="fresh"
)
last_refreshed_at: Mapped[Optional[datetime]] = mapped_column(
TIMESTAMP(timezone=True), nullable=True
)
error_code: Mapped[Optional[str]] = mapped_column(String(64), nullable=True)
error_message: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
created_at: Mapped[datetime] = mapped_column(
TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
)
updated_at: Mapped[datetime] = mapped_column(
TIMESTAMP(timezone=True), nullable=False, server_default=func.now()
)
__table_args__ = (
UniqueConstraint(
"connection_id",
"parent_ref",
name="uq_cloud_folder_states_connection_parent",
),
Index("ix_cloud_folder_states_connection", "connection_id"),
)
class Group(Base):
"""v2 stub — empty table, seeded for schema completeness (PROJECT.md D-02).
@@ -0,0 +1,208 @@
"""Add cloud_items, cloud_item_topics, and cloud_folder_states tables.
Revision ID: 0006
Revises: 0005
Create Date: 2026-06-18
Changes:
1. cloud_connections: add display_name_override (nullable Text) for user-defined
display names distinct from the auto-generated default. Does NOT add a
(user_id, provider) unique constraint — multiple same-provider accounts must
remain permitted (D-02, D-03).
2. cloud_items: durable per-item metadata indexed by connection/provider item ID.
No MinIO object_key field — provider bytes are never mirrored here.
analysis_status, semantic_index_status, semantic_index_data reserved for
Phase 14 analysis and semantic search.
3. cloud_item_topics: association between cloud items and existing owner topics.
No local Document row required.
4. cloud_folder_states: per-connection/parent-ref folder freshness row.
Root folder represented with parent_ref = '' (empty string, not NULL) to
allow a unique constraint on (connection_id, parent_ref).
Design notes:
- Ownership boundary: every cloud row has user_id + connection_id.
- Uniqueness: (connection_id, provider_item_id) for items; (connection_id, parent_ref)
for folder states. Same provider_item_id can coexist across connections.
- CASCADE: deleting a user or connection cascades all child rows automatically.
- D-18 compliance: no quota_used / byte_size field that flows to quotas.used_bytes.
"""
from __future__ import annotations
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import UUID as PG_UUID, JSONB
from alembic import op
# revision identifiers, used by Alembic.
revision = "0006"
down_revision = "0005"
branch_labels = None
depends_on = None
def upgrade() -> None:
# 1. Extend cloud_connections with optional user-defined display name override
op.add_column(
"cloud_connections",
sa.Column("display_name_override", sa.Text, nullable=True),
)
# 2. cloud_items table
op.create_table(
"cloud_items",
sa.Column(
"id",
PG_UUID(as_uuid=True),
primary_key=True,
server_default=sa.text("gen_random_uuid()"),
),
sa.Column(
"user_id",
PG_UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"connection_id",
PG_UUID(as_uuid=True),
sa.ForeignKey("cloud_connections.id", ondelete="CASCADE"),
nullable=False,
),
# Opaque provider-assigned identifier (e.g. Drive file ID, OneDrive driveItem id)
sa.Column("provider_item_id", sa.Text, nullable=False),
# Opaque reference to the parent folder (empty string = root)
sa.Column("parent_ref", sa.Text, nullable=True),
# Provider-reported path snapshot — informational only, not used for lookup
sa.Column("path_snapshot", sa.Text, nullable=True),
sa.Column("name", sa.Text, nullable=False),
# "file" | "folder"
sa.Column("kind", sa.String(8), nullable=False),
sa.Column("content_type", sa.Text, nullable=True),
# Provider-reported size — never flows to quotas.used_bytes (D-18)
sa.Column("provider_size", sa.BigInteger, nullable=True),
sa.Column("etag", sa.Text, nullable=True),
sa.Column("version", sa.Text, nullable=True),
sa.Column("modified_at", sa.TIMESTAMP(timezone=True), nullable=True),
sa.Column(
"last_seen_at",
sa.TIMESTAMP(timezone=True),
nullable=False,
server_default=sa.text("now()"),
),
sa.Column("deleted_at", sa.TIMESTAMP(timezone=True), nullable=True),
# Phase 14: byte analysis fields (reserved)
sa.Column("extracted_text", sa.Text, nullable=True),
# "pending" | "processing" | "done" | "error" | "skipped"
sa.Column("analysis_status", sa.String(16), nullable=False, server_default="pending"),
# "none" | "pending" | "indexed" | "error"
sa.Column("semantic_index_status", sa.String(16), nullable=False, server_default="none"),
# Provider-independent semantic search artifacts (reserved for Phase 15)
sa.Column("semantic_index_data", JSONB, nullable=True),
sa.Column(
"created_at",
sa.TIMESTAMP(timezone=True),
nullable=False,
server_default=sa.text("now()"),
),
sa.Column(
"updated_at",
sa.TIMESTAMP(timezone=True),
nullable=False,
server_default=sa.text("now()"),
),
# Uniqueness: same provider item ID can exist in different connections only
sa.UniqueConstraint(
"connection_id", "provider_item_id",
name="uq_cloud_items_connection_provider_item",
),
)
# Indexes for ownership lookups
op.create_index("ix_cloud_items_user_id", "cloud_items", ["user_id"])
op.create_index("ix_cloud_items_connection_id", "cloud_items", ["connection_id"])
op.create_index(
"ix_cloud_items_connection_parent",
"cloud_items",
["connection_id", "parent_ref"],
)
# 3. cloud_item_topics association table
op.create_table(
"cloud_item_topics",
sa.Column(
"cloud_item_id",
PG_UUID(as_uuid=True),
sa.ForeignKey("cloud_items.id", ondelete="CASCADE"),
primary_key=True,
),
sa.Column(
"topic_id",
PG_UUID(as_uuid=True),
sa.ForeignKey("topics.id", ondelete="CASCADE"),
primary_key=True,
),
)
# 4. cloud_folder_states table
# parent_ref uses empty string '' for root (allows unique constraint)
op.create_table(
"cloud_folder_states",
sa.Column(
"id",
PG_UUID(as_uuid=True),
primary_key=True,
server_default=sa.text("gen_random_uuid()"),
),
sa.Column(
"user_id",
PG_UUID(as_uuid=True),
sa.ForeignKey("users.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column(
"connection_id",
PG_UUID(as_uuid=True),
sa.ForeignKey("cloud_connections.id", ondelete="CASCADE"),
nullable=False,
),
# '' for root folder, provider ref string for non-root
sa.Column("parent_ref", sa.Text, nullable=False, server_default=""),
# "refreshing" | "fresh" | "warning"
sa.Column("refresh_state", sa.String(16), nullable=False, server_default="fresh"),
sa.Column("last_refreshed_at", sa.TIMESTAMP(timezone=True), nullable=True),
# Stable error code (e.g. "token_expired", "rate_limited") — never raw exception
sa.Column("error_code", sa.String(64), nullable=True),
# Safe human-readable message — never raw provider error text
sa.Column("error_message", sa.Text, nullable=True),
sa.Column(
"created_at",
sa.TIMESTAMP(timezone=True),
nullable=False,
server_default=sa.text("now()"),
),
sa.Column(
"updated_at",
sa.TIMESTAMP(timezone=True),
nullable=False,
server_default=sa.text("now()"),
),
sa.UniqueConstraint(
"connection_id", "parent_ref",
name="uq_cloud_folder_states_connection_parent",
),
)
op.create_index("ix_cloud_folder_states_connection", "cloud_folder_states", ["connection_id"])
def downgrade() -> None:
op.drop_index("ix_cloud_folder_states_connection", table_name="cloud_folder_states")
op.drop_table("cloud_folder_states")
op.drop_table("cloud_item_topics")
op.drop_index("ix_cloud_items_connection_parent", table_name="cloud_items")
op.drop_index("ix_cloud_items_connection_id", table_name="cloud_items")
op.drop_index("ix_cloud_items_user_id", table_name="cloud_items")
op.drop_table("cloud_items")
op.drop_column("cloud_connections", "display_name_override")
+282
View File
@@ -0,0 +1,282 @@
"""
Owner-scoped cloud metadata reconciliation service — Phase 12.
All functions operate within strict (user_id, connection_id) ownership boundaries.
No function here raises FastAPI HTTPException — domain exceptions only.
No function calls the quota service or alters quotas.used_bytes.
Domain exceptions:
ConnectionNotFound — connection does not exist or belongs to a different user.
CloudItemNotFound — cloud item does not exist for the given owner/connection.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Optional, Sequence
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from db.models import CloudConnection, CloudFolderState, CloudItem
from storage.cloud_base import CloudListing, CloudResource
# ── Domain exceptions ─────────────────────────────────────────────────────────
class ConnectionNotFound(ValueError):
"""Connection does not exist or belongs to a different user."""
class CloudItemNotFound(ValueError):
"""Cloud item does not exist for the given owner/connection."""
# ── Connection resolution ─────────────────────────────────────────────────────
async def resolve_owned_connection(
session: AsyncSession,
*,
connection_id: str,
user_id: str,
) -> CloudConnection:
"""Return the CloudConnection owned by user_id or raise ConnectionNotFound.
Never returns a connection belonging to another user.
"""
result = await session.execute(
select(CloudConnection).where(
CloudConnection.id == connection_id,
CloudConnection.user_id == user_id,
)
)
conn = result.scalars().first()
if conn is None:
raise ConnectionNotFound(
f"Connection {connection_id!r} not found for user {user_id!r}"
)
return conn
# ── Item listing ──────────────────────────────────────────────────────────────
async def list_cloud_children(
session: AsyncSession,
*,
user_id: str,
connection_id: str,
parent_ref: Optional[str],
) -> Sequence[CloudItem]:
"""Return non-deleted cloud items matching (user_id, connection_id, parent_ref).
parent_ref=None matches items whose parent_ref is NULL (root children where
parent is not tracked as a ref string).
"""
stmt = select(CloudItem).where(
CloudItem.user_id == user_id,
CloudItem.connection_id == connection_id,
CloudItem.deleted_at.is_(None),
)
if parent_ref is None:
stmt = stmt.where(CloudItem.parent_ref.is_(None))
else:
stmt = stmt.where(CloudItem.parent_ref == parent_ref)
result = await session.execute(stmt)
return result.scalars().all()
# ── Item upsert ───────────────────────────────────────────────────────────────
async def upsert_cloud_item(
session: AsyncSession,
*,
user_id: str,
resource: CloudResource,
) -> CloudItem:
"""Insert or update a CloudItem for the given normalized resource.
The DocuVault CloudItem UUID is preserved across provider rename/move:
if a row already exists for (connection_id, provider_item_id), its id
is retained and metadata fields are updated in place.
user_id is always set from the caller — never from the resource alone —
to enforce the owner boundary.
"""
connection_id = str(resource.connection_id)
result = await session.execute(
select(CloudItem).where(
CloudItem.connection_id == connection_id,
CloudItem.provider_item_id == resource.provider_item_id,
)
)
existing = result.scalars().first()
now = datetime.now(timezone.utc)
if existing is not None:
# Update metadata, preserve UUID (stable identity across rename/move)
existing.name = resource.name
existing.kind = resource.kind
existing.parent_ref = resource.parent_ref
existing.content_type = resource.content_type
existing.provider_size = resource.size
existing.etag = resource.etag
existing.modified_at = resource.modified_at
existing.last_seen_at = now
existing.deleted_at = None # un-delete if previously soft-deleted
existing.updated_at = now
await session.flush()
return existing
else:
item = CloudItem(
id=str(uuid.uuid4()),
user_id=user_id,
connection_id=connection_id,
provider_item_id=resource.provider_item_id,
parent_ref=resource.parent_ref,
name=resource.name,
kind=resource.kind,
content_type=resource.content_type,
provider_size=resource.size,
etag=resource.etag,
modified_at=resource.modified_at,
last_seen_at=now,
analysis_status="pending",
semantic_index_status="none",
)
session.add(item)
await session.flush()
return item
# ── Listing reconciliation ────────────────────────────────────────────────────
async def reconcile_cloud_listing(
session: AsyncSession,
*,
user_id: str,
connection_id: str,
parent_ref: Optional[str],
listing: CloudListing,
) -> None:
"""Reconcile a provider listing against durable cloud_items rows.
For each item in listing.items: upsert metadata.
Soft-deletion of missing children is ONLY performed when:
- listing.complete is True (authoritative, full listing)
Incomplete or failed listings (complete=False) must NEVER mark retained
rows deleted — provider bytes remain the source of truth.
This function never calls the quota service (D-18).
"""
seen_provider_ids: set[str] = set()
for resource in listing.items:
await upsert_cloud_item(session, user_id=user_id, resource=resource)
seen_provider_ids.add(resource.provider_item_id)
if listing.complete:
# Soft-delete items not present in the complete listing
stmt = select(CloudItem).where(
CloudItem.user_id == user_id,
CloudItem.connection_id == connection_id,
CloudItem.deleted_at.is_(None),
)
if parent_ref is None:
stmt = stmt.where(CloudItem.parent_ref.is_(None))
else:
stmt = stmt.where(CloudItem.parent_ref == parent_ref)
result = await session.execute(stmt)
existing_items = result.scalars().all()
now = datetime.now(timezone.utc)
for item in existing_items:
if item.provider_item_id not in seen_provider_ids:
item.deleted_at = now
item.updated_at = now
await session.flush()
# ── Folder state helpers ──────────────────────────────────────────────────────
async def get_or_create_folder_state(
session: AsyncSession,
*,
user_id: str,
connection_id: str,
parent_ref: str,
) -> CloudFolderState:
"""Return the CloudFolderState for (connection_id, parent_ref), creating if absent.
Idempotent: repeated calls with the same arguments return the same row.
parent_ref='' represents the connection root.
"""
result = await session.execute(
select(CloudFolderState).where(
CloudFolderState.connection_id == connection_id,
CloudFolderState.parent_ref == parent_ref,
)
)
existing = result.scalars().first()
if existing is not None:
return existing
fs = CloudFolderState(
id=str(uuid.uuid4()),
user_id=user_id,
connection_id=connection_id,
parent_ref=parent_ref,
refresh_state="fresh",
)
session.add(fs)
await session.flush()
return fs
async def update_folder_state(
session: AsyncSession,
*,
user_id: str,
connection_id: str,
parent_ref: str,
refresh_state: str,
last_refreshed_at: Optional[datetime] = None,
error_code: Optional[str] = None,
error_message: Optional[str] = None,
) -> CloudFolderState:
"""Update the refresh state for a folder.
Transitions: refreshing → fresh (success) or warning (failure).
On success: set last_refreshed_at.
On failure: retain last_refreshed_at, set error_code/message with controlled values.
error_code and error_message must be controlled service values — never raw
provider exception text.
"""
fs = await get_or_create_folder_state(
session, user_id=user_id, connection_id=connection_id, parent_ref=parent_ref
)
now = datetime.now(timezone.utc)
fs.refresh_state = refresh_state
fs.updated_at = now
if last_refreshed_at is not None:
fs.last_refreshed_at = last_refreshed_at
elif refresh_state == "fresh":
fs.last_refreshed_at = now
if refresh_state == "fresh":
fs.error_code = None
fs.error_message = None
elif error_code is not None:
fs.error_code = error_code
fs.error_message = error_message
await session.flush()
return fs
+245
View File
@@ -0,0 +1,245 @@
"""
Provider-neutral cloud resource capability contract for DocuVault Phase 12.
Defines the stable vocabulary (action keys, capability states, reason codes),
immutable normalized value types (CloudCapability, CloudResource, CloudListing),
and the abstract CloudResourceAdapter read-only interface.
Design decisions (12-CONTEXT.md D-06 through D-10):
- Structurally unsupported actions remain visible but greyed out (D-06).
- Capability discovery must never mutate provider content (D-08).
- Permanent vs. temporary limitations have distinct visual states (D-09).
- No mutation methods in Phase 12 contract; mutations added in Phase 13.
Credentials are never fields on any value type. The interface contains no
put, delete, rename, move, or create_folder methods.
"""
from __future__ import annotations
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
# ── Action keys ───────────────────────────────────────────────────────────────
ACTIONS = frozenset({
"browse",
"open",
"preview",
"upload",
"create_folder",
"rename",
"move",
"delete",
"change_tracking",
})
# ── Capability states ─────────────────────────────────────────────────────────
STATE_SUPPORTED = "supported"
STATE_UNSUPPORTED = "unsupported"
STATE_TEMPORARILY_UNAVAILABLE = "temporarily_unavailable"
CAPABILITY_STATES = frozenset({
STATE_SUPPORTED,
STATE_UNSUPPORTED,
STATE_TEMPORARILY_UNAVAILABLE,
})
# ── Stable reason codes ───────────────────────────────────────────────────────
REASON_PROVIDER_UNSUPPORTED = "provider_unsupported"
REASON_INSUFFICIENT_SCOPE = "insufficient_scope"
REASON_READ_ONLY = "read_only"
REASON_REAUTH_REQUIRED = "reauth_required"
REASON_OFFLINE = "offline"
REASON_ITEM_RESTRICTED = "item_restricted"
KNOWN_REASONS = frozenset({
REASON_PROVIDER_UNSUPPORTED,
REASON_INSUFFICIENT_SCOPE,
REASON_READ_ONLY,
REASON_REAUTH_REQUIRED,
REASON_OFFLINE,
REASON_ITEM_RESTRICTED,
})
# ── Normalized value types ────────────────────────────────────────────────────
@dataclass(frozen=True)
class CloudCapability:
"""Normalized capability descriptor for a single action.
state: one of the three CAPABILITY_STATES constants.
reason: one of the KNOWN_REASONS codes when state != supported; None otherwise.
message: controlled adapter output — safe for display; never raw provider text.
Invariants enforced at construction:
- state must be a known capability state.
- reason and message are only present when state is not supported.
- reason, when present, must be a known reason code.
"""
action: str
state: str
reason: Optional[str] = None
message: Optional[str] = None
def __post_init__(self) -> None:
if self.state not in CAPABILITY_STATES:
raise ValueError(
f"Unknown capability state {self.state!r}; "
f"must be one of {sorted(CAPABILITY_STATES)}"
)
if self.action not in ACTIONS:
raise ValueError(
f"Unknown action {self.action!r}; must be one of {sorted(ACTIONS)}"
)
if self.state == STATE_SUPPORTED:
if self.reason is not None or self.message is not None:
raise ValueError(
"reason and message must be None when state is 'supported'"
)
else:
if self.reason is not None and self.reason not in KNOWN_REASONS:
raise ValueError(
f"Unknown reason code {self.reason!r}; "
f"must be one of {sorted(KNOWN_REASONS)}"
)
@dataclass(frozen=True)
class CloudResource:
"""Normalized cloud item metadata — provider-independent representation.
id: stable DocuVault UUID for this item; survives provider rename/move.
provider_item_id: opaque provider-assigned identifier (e.g. Drive file ID).
connection_id: UUID of the CloudConnection that owns this item.
user_id: UUID of the owning user (security boundary).
name: display name as reported by the provider.
kind: "file" or "folder".
parent_ref: opaque provider reference to the parent folder; None at root.
content_type: MIME type for files; None for folders.
size: provider-reported size in bytes; None if not reported or for folders.
modified_at: provider-reported modification time; None if not available.
etag: provider etag or version string for change detection; None if absent.
capabilities: per-item capability overrides merged with connection defaults.
"""
id: uuid.UUID
provider_item_id: str
connection_id: uuid.UUID
user_id: uuid.UUID
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, CloudCapability] = field(default_factory=dict)
def __post_init__(self) -> None:
if self.kind not in ("file", "folder"):
raise ValueError(f"kind must be 'file' or 'folder', got {self.kind!r}")
for action, cap in self.capabilities.items():
if not isinstance(cap, CloudCapability):
raise TypeError(
f"capabilities[{action!r}] must be a CloudCapability instance"
)
@dataclass(frozen=True)
class CloudListing:
"""Result of listing a folder from a provider.
items: normalized CloudResource children of the listed folder.
complete: True when the listing represents the full authoritative contents
(safe to mark unseen children deleted on reconciliation).
False when the listing is partial, paginated, or failed — retained rows
must NOT be marked deleted.
next_page_token: opaque token for continuation; None when exhausted.
"""
items: tuple[CloudResource, ...]
complete: bool = True
next_page_token: Optional[str] = None
def __post_init__(self) -> None:
# Coerce list to tuple for immutability
object.__setattr__(self, "items", tuple(self.items))
# ── Abstract adapter ──────────────────────────────────────────────────────────
class CloudResourceAdapter(ABC):
"""Read-only cloud resource adapter contract for Phase 12.
Phase 12 interface is intentionally limited to browse and capability
discovery. No method here creates, renames, moves, or deletes provider
content. Mutation methods are added in Phase 13 on a separate subclass
interface.
Credentials are passed at construction time (or via dependency injection)
and are never exposed as fields or return values.
"""
@abstractmethod
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 the direct children of a folder.
parent_ref=None lists the connection root.
Returns a CloudListing with complete=False on partial or failed fetches.
Raises CloudConnectionError (from storage.exceptions) on auth failures.
Never mutates provider content as a side effect.
"""
...
@abstractmethod
async def get_capabilities(
self,
connection_id: uuid.UUID,
user_id: uuid.UUID,
) -> dict[str, CloudCapability]:
"""Return connection-level capabilities for all defined ACTIONS.
The returned dict must contain an entry for every action in ACTIONS.
Probing must not create, rename, move, or delete any provider content.
"""
...
def merge_item_capabilities(
self,
connection_caps: dict[str, CloudCapability],
item_caps: dict[str, CloudCapability],
) -> dict[str, CloudCapability]:
"""Merge per-item overrides onto connection defaults.
Item capabilities take precedence over connection capabilities.
Actions absent from item_caps fall back to connection_caps.
The result always contains every action in ACTIONS.
"""
merged: dict[str, CloudCapability] = {}
for action in ACTIONS:
if action in item_caps:
merged[action] = item_caps[action]
elif action in connection_caps:
merged[action] = connection_caps[action]
else:
# Safe default: unsupported with no reason
merged[action] = CloudCapability(
action=action,
state=STATE_UNSUPPORTED,
reason=REASON_PROVIDER_UNSUPPORTED,
message="This action is not supported by the provider.",
)
return merged
+330
View File
@@ -0,0 +1,330 @@
"""
Unit tests for the provider-neutral cloud resource capability contract.
Covers:
- Vocabulary completeness: all nine action keys and three capability states.
- Reason code completeness.
- CloudCapability validation (valid/invalid state, reason, messages).
- CloudResource construction and kind validation.
- CloudListing complete/incomplete semantics.
- Connection defaults, item overrides, and merge behavior.
- FakeAdapter: proves capability discovery invokes no put/delete/move/rename.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Optional
from unittest.mock import AsyncMock
import pytest
from storage.cloud_base import (
ACTIONS,
CAPABILITY_STATES,
KNOWN_REASONS,
STATE_SUPPORTED,
STATE_UNSUPPORTED,
STATE_TEMPORARILY_UNAVAILABLE,
REASON_PROVIDER_UNSUPPORTED,
REASON_INSUFFICIENT_SCOPE,
REASON_READ_ONLY,
REASON_REAUTH_REQUIRED,
REASON_OFFLINE,
REASON_ITEM_RESTRICTED,
CloudCapability,
CloudResource,
CloudListing,
CloudResourceAdapter,
)
# ── Vocabulary completeness ───────────────────────────────────────────────────
def test_actions_contains_all_nine():
expected = {
"browse", "open", "preview", "upload", "create_folder",
"rename", "move", "delete", "change_tracking",
}
assert ACTIONS == expected
def test_exactly_three_capability_states():
assert CAPABILITY_STATES == {
STATE_SUPPORTED,
STATE_UNSUPPORTED,
STATE_TEMPORARILY_UNAVAILABLE,
}
def test_known_reasons_all_present():
expected = {
REASON_PROVIDER_UNSUPPORTED,
REASON_INSUFFICIENT_SCOPE,
REASON_READ_ONLY,
REASON_REAUTH_REQUIRED,
REASON_OFFLINE,
REASON_ITEM_RESTRICTED,
}
assert KNOWN_REASONS == expected
# ── CloudCapability validation ────────────────────────────────────────────────
def test_supported_capability_no_reason_or_message():
cap = CloudCapability(action="browse", state=STATE_SUPPORTED)
assert cap.reason is None
assert cap.message is None
def test_supported_capability_with_reason_raises():
with pytest.raises(ValueError, match="reason and message must be None"):
CloudCapability(action="browse", state=STATE_SUPPORTED, reason=REASON_READ_ONLY)
def test_supported_capability_with_message_raises():
with pytest.raises(ValueError, match="reason and message must be None"):
CloudCapability(action="browse", state=STATE_SUPPORTED, message="some msg")
def test_unsupported_capability_preserves_reason_and_message():
cap = CloudCapability(
action="rename",
state=STATE_UNSUPPORTED,
reason=REASON_PROVIDER_UNSUPPORTED,
message="Rename is not supported by this server.",
)
assert cap.reason == REASON_PROVIDER_UNSUPPORTED
assert cap.message == "Rename is not supported by this server."
def test_temporarily_unavailable_with_remedy_message():
cap = CloudCapability(
action="upload",
state=STATE_TEMPORARILY_UNAVAILABLE,
reason=REASON_INSUFFICIENT_SCOPE,
message="Reconnect with write access to enable uploads.",
)
assert cap.state == STATE_TEMPORARILY_UNAVAILABLE
assert cap.message == "Reconnect with write access to enable uploads."
def test_unknown_state_raises():
with pytest.raises(ValueError, match="Unknown capability state"):
CloudCapability(action="browse", state="kinda_works")
def test_unknown_action_raises():
with pytest.raises(ValueError, match="Unknown action"):
CloudCapability(action="fly", state=STATE_SUPPORTED)
def test_unknown_reason_raises():
with pytest.raises(ValueError, match="Unknown reason code"):
CloudCapability(action="delete", state=STATE_UNSUPPORTED, reason="bad_reason")
def test_reason_none_for_unsupported_is_valid():
# reason is optional even for non-supported states
cap = CloudCapability(action="delete", state=STATE_UNSUPPORTED)
assert cap.reason is None
# ── CloudResource ─────────────────────────────────────────────────────────────
def _make_resource(**kwargs) -> CloudResource:
defaults = dict(
id=uuid.uuid4(),
provider_item_id="provider-abc",
connection_id=uuid.uuid4(),
user_id=uuid.uuid4(),
name="My Document.pdf",
kind="file",
)
defaults.update(kwargs)
return CloudResource(**defaults)
def test_resource_file_kind():
r = _make_resource(kind="file")
assert r.kind == "file"
def test_resource_folder_kind():
r = _make_resource(kind="folder")
assert r.kind == "folder"
def test_resource_invalid_kind_raises():
with pytest.raises(ValueError, match="kind must be"):
_make_resource(kind="symlink")
def test_resource_optional_fields_default_none():
r = _make_resource()
assert r.parent_ref is None
assert r.content_type is None
assert r.size is None
assert r.modified_at is None
assert r.etag is None
assert r.capabilities == {}
def test_resource_with_full_metadata():
now = datetime.now(timezone.utc)
cap = CloudCapability(action="delete", state=STATE_UNSUPPORTED, reason=REASON_ITEM_RESTRICTED)
r = _make_resource(
parent_ref="parent-001",
content_type="application/pdf",
size=102400,
modified_at=now,
etag="etag-v42",
capabilities={"delete": cap},
)
assert r.size == 102400
assert r.etag == "etag-v42"
assert r.capabilities["delete"] is cap
def test_resource_capability_invalid_type_raises():
with pytest.raises(TypeError):
_make_resource(capabilities={"delete": "not-a-capability"})
def test_resource_is_immutable():
r = _make_resource()
with pytest.raises(Exception): # dataclass frozen=True raises FrozenInstanceError
r.name = "changed" # type: ignore[misc]
# ── CloudListing ──────────────────────────────────────────────────────────────
def test_listing_complete_default():
listing = CloudListing(items=[])
assert listing.complete is True
def test_listing_items_coerced_to_tuple():
r = _make_resource()
listing = CloudListing(items=[r])
assert isinstance(listing.items, tuple)
assert listing.items[0] is r
def test_listing_incomplete_retains_rows():
r = _make_resource()
listing = CloudListing(items=[r], complete=False, next_page_token="tok-abc")
assert listing.complete is False
assert listing.next_page_token == "tok-abc"
# ── Merge behavior ────────────────────────────────────────────────────────────
class _FakeAdapter(CloudResourceAdapter):
"""Minimal test double that records which methods were called."""
def __init__(self):
self._calls: list[str] = []
async def list_folder(self, connection_id, user_id, parent_ref=None, page_token=None):
self._calls.append("list_folder")
return CloudListing(items=[], complete=True)
async def get_capabilities(self, connection_id, user_id):
self._calls.append("get_capabilities")
return {
action: CloudCapability(action=action, state=STATE_SUPPORTED)
for action in ACTIONS
}
# Explicitly absent: any method that could mutate provider state.
# Python raises AttributeError naturally if called — confirmed below.
def test_fake_adapter_has_no_mutation_methods():
"""Prove the Phase 12 interface defines no put/delete/move/rename methods."""
adapter = _FakeAdapter()
mutation_methods = ["put_object", "delete_object", "rename_item", "move_item", "create_folder_remote"]
for method_name in mutation_methods:
assert not hasattr(adapter, method_name), (
f"Phase 12 adapter must not expose mutation method {method_name!r}"
)
@pytest.mark.asyncio
async def test_fake_adapter_list_folder_invokes_no_mutation():
adapter = _FakeAdapter()
cid = uuid.uuid4()
uid = uuid.uuid4()
listing = await adapter.list_folder(cid, uid)
assert isinstance(listing, CloudListing)
assert "list_folder" in adapter._calls
# Only list_folder was called, no mutation methods
assert adapter._calls == ["list_folder"]
@pytest.mark.asyncio
async def test_fake_adapter_get_capabilities_covers_all_actions():
adapter = _FakeAdapter()
cid = uuid.uuid4()
uid = uuid.uuid4()
caps = await adapter.get_capabilities(cid, uid)
for action in ACTIONS:
assert action in caps, f"Missing capability for action {action!r}"
assert isinstance(caps[action], CloudCapability)
def test_merge_item_override_takes_precedence():
adapter = _FakeAdapter()
connection_caps = {
action: CloudCapability(action=action, state=STATE_SUPPORTED)
for action in ACTIONS
}
item_override = {
"delete": CloudCapability(
action="delete",
state=STATE_UNSUPPORTED,
reason=REASON_ITEM_RESTRICTED,
message="This item cannot be deleted.",
)
}
merged = adapter.merge_item_capabilities(connection_caps, item_override)
assert merged["delete"].state == STATE_UNSUPPORTED
assert merged["delete"].reason == REASON_ITEM_RESTRICTED
# Non-overridden actions fall back to connection caps
assert merged["browse"].state == STATE_SUPPORTED
def test_merge_covers_all_actions():
adapter = _FakeAdapter()
merged = adapter.merge_item_capabilities({}, {})
for action in ACTIONS:
assert action in merged
def test_merge_empty_item_caps_uses_connection_defaults():
adapter = _FakeAdapter()
connection_caps = {
action: CloudCapability(action=action, state=STATE_SUPPORTED)
for action in ACTIONS
}
merged = adapter.merge_item_capabilities(connection_caps, {})
for action in ACTIONS:
assert merged[action].state == STATE_SUPPORTED
def test_exact_reason_and_message_preserved_through_merge():
adapter = _FakeAdapter()
cap = CloudCapability(
action="rename",
state=STATE_TEMPORARILY_UNAVAILABLE,
reason=REASON_REAUTH_REQUIRED,
message="Re-authenticate to rename files.",
)
merged = adapter.merge_item_capabilities(
{action: CloudCapability(action=action, state=STATE_SUPPORTED) for action in ACTIONS},
{"rename": cap},
)
result = merged["rename"]
assert result.reason == REASON_REAUTH_REQUIRED
assert result.message == "Re-authenticate to rename files."
+528
View File
@@ -0,0 +1,528 @@
"""
Tests for cloud item metadata schema, model isolation, and reconciliation service.
Task 2 tests (model/schema/isolation/quota):
- ORM model fields match migration schema expectations.
- Metadata, extracted text, topic links, and semantic data persist without
requiring an object key or byte cache row.
- Identical provider item IDs coexist across different connections.
- Foreign-owner queries do not return rows for other users.
- Provider size never alters Quota.used_bytes.
- Root folder state can be represented without a CloudItem parent row.
Task 3 tests (reconciliation service):
- Repeated reconciliation creates no duplicate rows.
- Rename/move updates metadata without changing the CloudItem UUID.
- Incomplete or failed listing cannot mark unseen children deleted.
- All queries include both owner and connection scope.
- Idempotent repeated reconciliation.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from typing import Optional
import pytest
import pytest_asyncio
from sqlalchemy import String, 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 db.models import Base, CloudItem, CloudItemTopic, CloudFolderState, CloudConnection, Topic, User, Quota
from storage.cloud_base import CloudListing, CloudResource, CloudCapability, STATE_SUPPORTED
from services.cloud_items import (
resolve_owned_connection,
list_cloud_children,
upsert_cloud_item,
reconcile_cloud_listing,
get_or_create_folder_state,
update_folder_state,
CloudItemNotFound,
ConnectionNotFound,
)
# ── SQLite test engine setup ──────────────────────────────────────────────────
@pytest_asyncio.fixture
async def db_session():
"""In-memory SQLite session with PostgreSQL-type shims."""
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
@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()
# ── Helpers ───────────────────────────────────────────────────────────────────
def _user_id() -> str:
return str(uuid.uuid4())
def _conn_id() -> str:
return str(uuid.uuid4())
def _item_id() -> str:
return str(uuid.uuid4())
async def _make_user(session: AsyncSession, user_id: Optional[str] = None) -> str:
uid = user_id or _user_id()
u = User(
id=uid,
handle=f"user_{uid[:8]}",
email=f"{uid[:8]}@example.com",
password_hash="hash",
role="user",
)
session.add(u)
await session.flush()
return uid
async def _make_connection(
session: AsyncSession, user_id: str, conn_id: Optional[str] = None
) -> str:
cid = conn_id or _conn_id()
conn = CloudConnection(
id=cid,
user_id=user_id,
provider="onedrive",
display_name="My OneDrive",
credentials_enc="enc",
status="ACTIVE",
)
session.add(conn)
await session.flush()
return cid
def _cloud_resource(
connection_id: str,
user_id: str,
provider_item_id: str = "item-001",
name: str = "test.pdf",
kind: str = "file",
parent_ref: Optional[str] = None,
size: Optional[int] = None,
etag: Optional[str] = None,
) -> CloudResource:
return CloudResource(
id=uuid.uuid4(),
provider_item_id=provider_item_id,
connection_id=uuid.UUID(connection_id),
user_id=uuid.UUID(user_id),
name=name,
kind=kind,
parent_ref=parent_ref,
size=size,
etag=etag,
)
# ── Task 2: model/schema tests ────────────────────────────────────────────────
@pytest.mark.asyncio
async def test_model_cloud_item_basic_fields(db_session: AsyncSession):
"""CloudItem persists metadata without requiring an object key."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
item_id = _item_id()
item = CloudItem(
id=item_id,
user_id=uid,
connection_id=cid,
provider_item_id="driveitem-123",
name="report.pdf",
kind="file",
content_type="application/pdf",
provider_size=204800,
etag="etag-v1",
)
db_session.add(item)
await db_session.flush()
await db_session.refresh(item)
assert item.id == item_id
assert item.provider_item_id == "driveitem-123"
assert item.name == "report.pdf"
assert item.kind == "file"
assert item.provider_size == 204800
assert item.analysis_status == "pending"
assert item.semantic_index_status == "none"
assert item.deleted_at is None
@pytest.mark.asyncio
async def test_model_cloud_item_extracted_text_without_object_key(db_session: AsyncSession):
"""Extracted text and semantic data persist with no MinIO object key field."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
item = CloudItem(
id=_item_id(),
user_id=uid,
connection_id=cid,
provider_item_id="item-text",
name="notes.txt",
kind="file",
extracted_text="Sample extracted text",
semantic_index_status="indexed",
# No object_key field exists — provider bytes are not retained here
)
db_session.add(item)
await db_session.flush()
await db_session.refresh(item)
assert item.extracted_text == "Sample extracted text"
assert item.semantic_index_status == "indexed"
# Confirm no object_key attribute at all
assert not hasattr(item, "object_key")
@pytest.mark.asyncio
async def test_model_cloud_item_topics_no_document_row(db_session: AsyncSession):
"""CloudItemTopic links item to topic without requiring a Document row."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
topic = Topic(id=_item_id(), user_id=uid, name="Finance", description="", color="#ff0000")
db_session.add(topic)
await db_session.flush()
item = CloudItem(
id=_item_id(),
user_id=uid,
connection_id=cid,
provider_item_id="item-topic",
name="budget.xlsx",
kind="file",
)
db_session.add(item)
await db_session.flush()
link = CloudItemTopic(cloud_item_id=item.id, topic_id=topic.id)
db_session.add(link)
await db_session.flush()
# Verify link persisted
from sqlalchemy import select
result = await db_session.execute(
select(CloudItemTopic).where(CloudItemTopic.cloud_item_id == item.id)
)
rows = result.scalars().all()
assert len(rows) == 1
assert rows[0].topic_id == topic.id
@pytest.mark.asyncio
async def test_model_schema_isolation_same_provider_item_different_connections(db_session: AsyncSession):
"""Identical provider_item_id can coexist across different connections."""
uid = await _make_user(db_session)
cid1 = await _make_connection(db_session, uid)
cid2 = await _make_connection(db_session, uid)
shared_pid = "shared-item-001"
item1 = CloudItem(
id=_item_id(), user_id=uid, connection_id=cid1,
provider_item_id=shared_pid, name="file.pdf", kind="file",
)
item2 = CloudItem(
id=_item_id(), user_id=uid, connection_id=cid2,
provider_item_id=shared_pid, name="file.pdf", kind="file",
)
db_session.add_all([item1, item2])
await db_session.flush()
from sqlalchemy import select
result = await db_session.execute(
select(CloudItem).where(CloudItem.provider_item_id == shared_pid)
)
rows = result.scalars().all()
assert len(rows) == 2
@pytest.mark.asyncio
async def test_model_schema_foreign_owner_query_returns_nothing(db_session: AsyncSession):
"""Items for user A are not returned when querying for user B."""
uid_a = await _make_user(db_session)
uid_b = await _make_user(db_session)
cid_a = await _make_connection(db_session, uid_a)
item = CloudItem(
id=_item_id(), user_id=uid_a, connection_id=cid_a,
provider_item_id="item-a", name="private.pdf", kind="file",
)
db_session.add(item)
await db_session.flush()
from sqlalchemy import select
result = await db_session.execute(
select(CloudItem).where(
CloudItem.user_id == uid_b,
CloudItem.provider_item_id == "item-a",
)
)
assert result.scalars().first() is None
@pytest.mark.asyncio
async def test_model_quota_unchanged_after_item_upsert(db_session: AsyncSession):
"""Upserting a cloud item with provider_size must not alter Quota.used_bytes."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
quota = Quota(user_id=uid, limit_bytes=104857600, used_bytes=0)
db_session.add(quota)
await db_session.flush()
item = CloudItem(
id=_item_id(), user_id=uid, connection_id=cid,
provider_item_id="large-file", name="video.mp4", kind="file",
provider_size=5_000_000_000, # 5 GB provider-reported size
)
db_session.add(item)
await db_session.flush()
await db_session.refresh(quota)
# Quota must remain unchanged
assert quota.used_bytes == 0
@pytest.mark.asyncio
async def test_model_folder_state_root_no_parent_item(db_session: AsyncSession):
"""Root folder state can be created with parent_ref='' without a CloudItem row."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
fs = CloudFolderState(
id=_item_id(),
user_id=uid,
connection_id=cid,
parent_ref="", # root
refresh_state="fresh",
)
db_session.add(fs)
await db_session.flush()
await db_session.refresh(fs)
assert fs.parent_ref == ""
assert fs.refresh_state == "fresh"
assert fs.last_refreshed_at is None
# ── Task 3: reconciliation service tests ─────────────────────────────────────
@pytest.mark.asyncio
async def test_service_resolve_owned_connection_found(db_session: AsyncSession):
uid = await _make_user(db_session)
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
@pytest.mark.asyncio
async def test_service_resolve_owned_connection_wrong_owner_raises(db_session: AsyncSession):
uid_a = await _make_user(db_session)
uid_b = await _make_user(db_session)
cid = await _make_connection(db_session, uid_a)
with pytest.raises(ConnectionNotFound):
await resolve_owned_connection(db_session, connection_id=cid, user_id=uid_b)
@pytest.mark.asyncio
async def test_service_upsert_creates_item(db_session: AsyncSession):
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
resource = _cloud_resource(cid, uid, provider_item_id="new-001", name="doc.pdf")
item = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
assert item.provider_item_id == "new-001"
assert item.name == "doc.pdf"
@pytest.mark.asyncio
async def test_service_upsert_idempotent_same_uuid(db_session: AsyncSession):
"""Repeated upsert with unchanged data creates no duplicate row."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
resource = _cloud_resource(cid, uid, provider_item_id="idem-001")
item1 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
item2 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
assert item1.id == item2.id # same DocuVault UUID preserved
@pytest.mark.asyncio
async def test_service_upsert_rename_preserves_uuid(db_session: AsyncSession):
"""Rename updates name but keeps the same CloudItem UUID."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
resource = _cloud_resource(cid, uid, provider_item_id="rename-001", name="old.pdf")
item1 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
original_uuid = item1.id
renamed = _cloud_resource(cid, uid, provider_item_id="rename-001", name="new.pdf")
item2 = await upsert_cloud_item(db_session, user_id=uid, resource=renamed)
assert item2.id == original_uuid
assert item2.name == "new.pdf"
@pytest.mark.asyncio
async def test_service_list_children_owner_scoped(db_session: AsyncSession):
"""list_cloud_children only returns items for the correct (user, connection, parent)."""
uid_a = await _make_user(db_session)
uid_b = await _make_user(db_session)
cid_a = await _make_connection(db_session, uid_a)
res = _cloud_resource(cid_a, uid_a, provider_item_id="child-001", parent_ref="parent-ref")
await upsert_cloud_item(db_session, user_id=uid_a, resource=res)
# User A sees the item
children_a = await list_cloud_children(
db_session, user_id=uid_a, connection_id=cid_a, parent_ref="parent-ref"
)
assert len(children_a) == 1
# User B does not
children_b = await list_cloud_children(
db_session, user_id=uid_b, connection_id=cid_a, parent_ref="parent-ref"
)
assert len(children_b) == 0
@pytest.mark.asyncio
async def test_service_reconcile_complete_marks_missing_deleted(db_session: AsyncSession):
"""Complete listing marks items not in the listing as deleted."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
# Pre-existing item
existing = _cloud_resource(cid, uid, provider_item_id="keep-001", parent_ref="root")
gone = _cloud_resource(cid, uid, provider_item_id="gone-001", parent_ref="root")
await upsert_cloud_item(db_session, user_id=uid, resource=existing)
await upsert_cloud_item(db_session, user_id=uid, resource=gone)
# Reconcile with only "keep-001" in the complete listing
listing = CloudListing(items=[existing], complete=True)
await reconcile_cloud_listing(
db_session, user_id=uid, connection_id=cid, parent_ref="root", listing=listing
)
from sqlalchemy import select
result = await db_session.execute(
select(CloudItem).where(CloudItem.provider_item_id == "gone-001")
)
gone_item = result.scalars().first()
assert gone_item is not None
assert gone_item.deleted_at is not None
@pytest.mark.asyncio
async def test_service_reconcile_incomplete_does_not_delete(db_session: AsyncSession):
"""Incomplete (partial/failed) listing must not mark unseen children deleted."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
item = _cloud_resource(cid, uid, provider_item_id="retain-001", parent_ref="root")
await upsert_cloud_item(db_session, user_id=uid, resource=item)
# Reconcile with empty incomplete listing
listing = CloudListing(items=[], complete=False)
await reconcile_cloud_listing(
db_session, user_id=uid, connection_id=cid, parent_ref="root", listing=listing
)
from sqlalchemy import select
result = await db_session.execute(
select(CloudItem).where(CloudItem.provider_item_id == "retain-001")
)
retained = result.scalars().first()
assert retained is not None
assert retained.deleted_at is None
@pytest.mark.asyncio
async def test_service_folder_state_create_and_update(db_session: AsyncSession):
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
fs = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
assert fs.refresh_state == "fresh"
await update_folder_state(
db_session,
user_id=uid,
connection_id=cid,
parent_ref="",
refresh_state="warning",
error_code="token_expired",
error_message="Re-authenticate to refresh.",
)
from sqlalchemy import select
result = await db_session.execute(
select(CloudFolderState).where(
CloudFolderState.connection_id == cid,
CloudFolderState.parent_ref == "",
)
)
updated = result.scalars().first()
assert updated.refresh_state == "warning"
assert updated.error_code == "token_expired"
@pytest.mark.asyncio
async def test_service_folder_state_idempotent(db_session: AsyncSession):
"""get_or_create is idempotent — no duplicate rows."""
uid = await _make_user(db_session)
cid = await _make_connection(db_session, uid)
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