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

feat(12-01): normalized cloud resource capability contract and unit tests

Browse Source 
- Define 9 action keys, 3 capability states, 6 reason codes in cloud_base.py
- Immutable CloudCapability, CloudResource, CloudListing frozen dataclasses
- 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, fake adapter
This commit is contained in:
curo1305
2026-06-18 22:34:23 +02:00
parent 11b91775b6
commit 0a7273b9fe
2 changed files with 575 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/storage/cloud_base.py
+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
backend/tests/test_cloud_capabilities.py
+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."