curo1305
0a7273b9fe
- 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
331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""
|
|
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."
|