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

test(13-01): extend provider contract suites for four-provider mutable-operation parity

Browse Source 
- Add Phase 13 mutable-operation RED tests to test_cloud_backends.py:
  TestGoogleDriveMutableContract (D-17 scope, create/rename/move/delete/upload),
  TestOneDriveMutableContract (CONN-02 token handoff, permanent-delete disclosure,
  nextLink SSRF guard), TestNextcloudMutableContract (create-folder SSRF, delete
  normalization), TestWebDAVMutableContract (permanent-delete, move SSRF, no
  cloud_items imports)
- Add Phase 13 mutable-operation RED tests to test_cloud_provider_contract.py:
  TestMutableAdapterContract (method existence, async contract, canonical signatures
  for all four providers), TestMutableAdapterResultNormalization (normalized kind/reason
  return types, conflict normalization documentation, unsupported-capability disclosure)
- All new tests fail against the current codebase — mutable adapter methods do not
  exist yet (expected RED); all prior Phase 12 tests remain green
This commit is contained in:
curo1305
2026-06-22 18:01:15 +02:00
parent efb596433c
commit fd6b561899
2 changed files with 978 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_cloud_backends.py
+682
View File
@@ -25,6 +25,8 @@ from __future__ import annotations
import asyncio
import inspect
import uuid as _uuid
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@@ -795,3 +797,683 @@ class TestWebDAVListFolder:
assert set(caps.keys()) == ACTIONS
assert caps["browse"].state == "supported"
# =============================================================================
# Phase 13 Plan 01 — RED: Provider-specific mutable-operation contracts
#
# All tests below FAIL until Phase 13 implements the mutable adapter interface.
# =============================================================================
class TestGoogleDriveMutableContract:
"""Google Drive — Phase 13 mutable-operation contract (RED).
D-17: Phase 13 requests broader drive.files scope — test must assert
the consent scope includes 'drive' or 'drive.file' as the minimum for
listing and operating on pre-existing Drive items.
All method assertions will fail until MutableCloudResourceAdapter is
implemented in Phase 13.
"""
def _make_backend(self):
from storage.google_drive_backend import GoogleDriveBackend
return GoogleDriveBackend({
"access_token": "ya29.test_13",
"refresh_token": "1//test_refresh_13",
"token_uri": "https://oauth2.googleapis.com/token",
"client_id": "test_client_13",
"client_secret": "test_secret_13",
})
def test_create_folder_is_async(self):
"""create_folder must be an async coroutine on GoogleDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.google_drive_backend import GoogleDriveBackend
assert hasattr(GoogleDriveBackend, "create_folder"), (
"GoogleDriveBackend must implement create_folder (Phase 13)"
)
assert inspect.iscoroutinefunction(GoogleDriveBackend.create_folder), (
"create_folder must be async"
)
def test_rename_is_async(self):
"""rename must be an async coroutine on GoogleDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.google_drive_backend import GoogleDriveBackend
assert hasattr(GoogleDriveBackend, "rename"), (
"GoogleDriveBackend must implement rename (Phase 13)"
)
assert inspect.iscoroutinefunction(GoogleDriveBackend.rename)
def test_move_is_async(self):
"""move must be an async coroutine on GoogleDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.google_drive_backend import GoogleDriveBackend
assert hasattr(GoogleDriveBackend, "move"), (
"GoogleDriveBackend must implement move (Phase 13)"
)
assert inspect.iscoroutinefunction(GoogleDriveBackend.move)
def test_delete_is_async(self):
"""delete must be an async coroutine on GoogleDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.google_drive_backend import GoogleDriveBackend
assert hasattr(GoogleDriveBackend, "delete"), (
"GoogleDriveBackend must implement delete (Phase 13)"
)
assert inspect.iscoroutinefunction(GoogleDriveBackend.delete)
def test_upload_file_is_async(self):
"""upload_file must be an async coroutine on GoogleDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.google_drive_backend import GoogleDriveBackend
assert hasattr(GoogleDriveBackend, "upload_file"), (
"GoogleDriveBackend must implement upload_file (Phase 13)"
)
assert inspect.iscoroutinefunction(GoogleDriveBackend.upload_file)
@pytest.mark.asyncio
async def test_delete_prefers_trash_normalizes_result(self):
"""Google Drive delete should prefer Trash and return typed result dict.
D-11: Provider trash is preferred over permanent delete. The normalized
result must carry {'kind': 'deleted', 'reason': 'trashed'} or 'permanent'.
FAILS: delete method does not exist yet.
"""
from storage.google_drive_backend import GoogleDriveBackend
backend = self._make_backend()
fake_service = MagicMock()
fake_service.files().trash().execute.return_value = {}
backend._get_service = MagicMock(return_value=fake_service)
result = await backend.delete("file_id_123", trash=True)
assert isinstance(result, dict), "delete must return a normalized dict result"
assert result.get("kind") == "deleted", (
f"Expected kind='deleted', got {result.get('kind')!r}"
)
assert result.get("reason") in ("trashed", "permanent"), (
f"Expected reason 'trashed' or 'permanent', got {result.get('reason')!r}"
)
@pytest.mark.asyncio
async def test_create_folder_name_collision_normalizes_result(self):
"""create_folder with a collision returns normalized conflict kind/reason.
D-05: Create-folder collision must return {'kind': 'conflict', 'reason': 'name_collision'}
rather than raising an unhandled API error.
FAILS: create_folder method does not exist yet.
"""
from storage.google_drive_backend import GoogleDriveBackend
backend = self._make_backend()
fake_service = MagicMock()
# Simulate Drive returning a 409 or duplicate error
from googleapiclient.errors import HttpError
fake_resp = MagicMock()
fake_resp.status = 409
fake_service.files().create().execute.side_effect = HttpError(
resp=fake_resp, content=b'{"error":{"message":"File already exists"}}'
)
backend._get_service = MagicMock(return_value=fake_service)
result = await backend.create_folder(
parent_ref=None,
name="Existing Folder",
connection_id=_uuid.uuid4(),
user_id=_uuid.uuid4(),
)
assert isinstance(result, dict), "create_folder must return a normalized dict"
assert result.get("kind") == "conflict", (
f"Expected kind='conflict', got {result.get('kind')!r}"
)
@pytest.mark.asyncio
async def test_rename_stale_metadata_normalizes_result(self):
"""rename with mismatched etag returns normalized stale kind/reason.
D-07: Stale rename must return {'kind': 'stale', 'reason': 'item_changed'}.
FAILS: rename method does not exist yet.
"""
from storage.google_drive_backend import GoogleDriveBackend
backend = self._make_backend()
fake_service = MagicMock()
from googleapiclient.errors import HttpError
fake_resp = MagicMock()
fake_resp.status = 412
fake_service.files().update().execute.side_effect = HttpError(
resp=fake_resp, content=b'{"error":{"message":"precondition failed"}}'
)
backend._get_service = MagicMock(return_value=fake_service)
result = await backend.rename(
provider_item_id="file_id_123",
new_name="Renamed.pdf",
etag="old-etag",
)
assert isinstance(result, dict), "rename must return a normalized dict"
assert result.get("kind") == "stale", (
f"Expected kind='stale', got {result.get('kind')!r}"
)
def test_drive_scope_includes_full_drive_access(self):
"""Google Drive backend must request 'drive' or 'drive.file' scope (D-17).
D-17: Phase 13 requests broader Google Drive access — 'drive.file' scope only
allows access to files created by the app. Full drive browsing requires 'drive'.
FAILS: Phase 13 scope expansion not yet implemented.
"""
import inspect
import re
from storage.google_drive_backend import GoogleDriveBackend
# The backend's OAuth scope configuration must include full drive access
# Check for SCOPES constant or _SCOPES attribute
scopes = getattr(GoogleDriveBackend, "SCOPES", None) or \
getattr(GoogleDriveBackend, "_SCOPES", None) or \
getattr(GoogleDriveBackend, "OAUTH_SCOPES", None)
assert scopes is not None, (
"GoogleDriveBackend must declare a SCOPES / _SCOPES / OAUTH_SCOPES class attribute "
"for the OAuth consent scope (D-17)"
)
scope_str = " ".join(scopes) if isinstance(scopes, (list, tuple)) else str(scopes)
# Phase 13 requires at minimum 'drive' scope (not just 'drive.file' or 'drive.readonly')
assert "https://www.googleapis.com/auth/drive" in scope_str or "drive" in scope_str, (
f"GoogleDriveBackend SCOPES must include full Drive access for Phase 13 (D-17), "
f"got: {scope_str!r}"
)
class TestOneDriveMutableContract:
"""OneDrive — Phase 13 mutable-operation contract (RED).
CONN-02: Successful token refresh must hand the new access_token and
refresh_token off to the service layer for encrypted persistence.
All method assertions will fail until Phase 13 implements the mutable interface.
"""
def _make_backend(self, expires_at: str = "2099-01-01T00:00:00"):
from storage.onedrive_backend import OneDriveBackend
return OneDriveBackend({
"access_token": "od_access_13",
"refresh_token": "od_refresh_13",
"expires_at": expires_at,
})
def test_create_folder_is_async(self):
"""create_folder must be an async coroutine on OneDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.onedrive_backend import OneDriveBackend
assert hasattr(OneDriveBackend, "create_folder"), (
"OneDriveBackend must implement create_folder (Phase 13)"
)
assert inspect.iscoroutinefunction(OneDriveBackend.create_folder)
def test_rename_is_async(self):
"""rename must be an async coroutine on OneDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.onedrive_backend import OneDriveBackend
assert hasattr(OneDriveBackend, "rename"), (
"OneDriveBackend must implement rename (Phase 13)"
)
assert inspect.iscoroutinefunction(OneDriveBackend.rename)
def test_move_is_async(self):
"""move must be an async coroutine on OneDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.onedrive_backend import OneDriveBackend
assert hasattr(OneDriveBackend, "move"), (
"OneDriveBackend must implement move (Phase 13)"
)
assert inspect.iscoroutinefunction(OneDriveBackend.move)
def test_delete_is_async(self):
"""delete must be an async coroutine on OneDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.onedrive_backend import OneDriveBackend
assert hasattr(OneDriveBackend, "delete"), (
"OneDriveBackend must implement delete (Phase 13)"
)
assert inspect.iscoroutinefunction(OneDriveBackend.delete)
def test_upload_file_is_async(self):
"""upload_file must be an async coroutine on OneDriveBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.onedrive_backend import OneDriveBackend
assert hasattr(OneDriveBackend, "upload_file"), (
"OneDriveBackend must implement upload_file (Phase 13)"
)
assert inspect.iscoroutinefunction(OneDriveBackend.upload_file)
@pytest.mark.asyncio
async def test_refresh_token_hands_off_new_credentials(self):
"""Successful token refresh returns new credentials for service-layer persistence.
CONN-02: _refresh_token must return the new credentials dict (or signal the
service layer) so encrypted persistence can happen. The adapter itself must
not encrypt or store — that is the service layer's responsibility.
FAILS: _refresh_token return contract not yet established.
"""
from storage.onedrive_backend import OneDriveBackend
backend = self._make_backend(expires_at="2000-01-01T00:00:00") # expired
new_access = "new_access_token_13"
new_refresh = "new_refresh_token_13"
import msal
fake_app = MagicMock()
fake_app.acquire_token_by_refresh_token.return_value = {
"access_token": new_access,
"refresh_token": new_refresh,
"expires_in": 3600,
}
with patch("msal.PublicClientApplication", return_value=fake_app):
refreshed = await backend._refresh_token()
# _refresh_token must return the new credentials dict for service-layer handoff
assert refreshed is not None, (
"_refresh_token must return new credentials dict for CONN-02 persistence handoff"
)
assert isinstance(refreshed, dict), (
"_refresh_token must return a dict of new credentials"
)
assert refreshed.get("access_token") == new_access, (
f"Refreshed access_token mismatch: expected {new_access!r}, got {refreshed.get('access_token')!r}"
)
@pytest.mark.asyncio
async def test_delete_permanent_normalizes_result(self):
"""OneDrive delete returns normalized {'kind': 'deleted', 'reason': 'permanent'}.
D-11: OneDrive does not have a REST trash API — deletes are permanent.
The result must make this explicit to the service layer / frontend.
FAILS: delete method does not exist yet.
"""
from storage.onedrive_backend import OneDriveBackend
backend = self._make_backend()
mock_resp = MagicMock()
mock_resp.status_code = 204
mock_resp.is_success = True
with patch("httpx.AsyncClient") as mock_cls:
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=None)
mock_client.delete = AsyncMock(return_value=mock_resp)
mock_cls.return_value = mock_client
result = await backend.delete("od_item_id_123")
assert isinstance(result, dict), "delete must return a normalized dict"
assert result.get("kind") == "deleted"
assert result.get("reason") == "permanent", (
"OneDrive deletes are permanent — reason must be 'permanent' (D-11)"
)
@pytest.mark.asyncio
async def test_nextlink_restricted_to_graph_host(self):
"""OneDrive nextLink must only be followed to graph.microsoft.com host (SSRF guard).
An SSRF attack could inject a nextLink pointing to an internal IP or
attacker-controlled server. The backend must restrict nextLink to the
known Graph API host.
FAILS: Phase 13 nextLink host guard not yet implemented.
"""
from storage.onedrive_backend import OneDriveBackend
import uuid
backend = self._make_backend()
# Inject a malicious nextLink pointing to an attacker host
malicious_page1 = {
"value": [{"id": "od1", "name": "file.txt", "file": {}}],
"@odata.nextLink": "http://169.254.169.254/latest/meta-data/",
}
mock_resp = MagicMock()
mock_resp.is_success = True
mock_resp.json.return_value = malicious_page1
with patch("httpx.AsyncClient") as mock_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_cls.return_value = mock_client
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
# The backend must NOT follow the malicious nextLink
# Result should be complete=False (truncated at the SSRF boundary)
# OR the listing should be returned without following the malicious page
assert mock_client.get.call_count >= 1
# Verify the second call (if made) went only to graph.microsoft.com
if mock_client.get.call_count > 1:
for call_args in mock_client.get.call_args_list[1:]:
url = call_args[0][0] if call_args[0] else call_args[1].get("url", "")
assert "graph.microsoft.com" in url or "microsoftonline" in url, (
f"nextLink follow must be restricted to Graph host — SSRF guard (T-13-04); "
f"got URL: {url!r}"
)
class TestNextcloudMutableContract:
"""Nextcloud — Phase 13 mutable-operation contract (RED).
T-13-04: Nextcloud/WebDAV outbound calls must be SSRF-guarded.
All method assertions will fail until Phase 13 implements the mutable interface.
"""
def _make_backend(self):
from storage.nextcloud_backend import NextcloudBackend
with patch("storage.webdav_backend.validate_cloud_url"):
with patch("webdav3.client.Client"):
backend = NextcloudBackend.__new__(NextcloudBackend)
backend._server_url = "https://nc.example.com/remote.php/dav/files/user/"
backend._username = "testuser"
backend._client = MagicMock()
return backend
def test_create_folder_is_async(self):
"""create_folder must be an async coroutine on NextcloudBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.nextcloud_backend import NextcloudBackend
assert hasattr(NextcloudBackend, "create_folder"), (
"NextcloudBackend must implement create_folder (Phase 13)"
)
assert inspect.iscoroutinefunction(NextcloudBackend.create_folder)
def test_rename_is_async(self):
"""rename must be an async coroutine on NextcloudBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.nextcloud_backend import NextcloudBackend
assert hasattr(NextcloudBackend, "rename"), (
"NextcloudBackend must implement rename (Phase 13)"
)
assert inspect.iscoroutinefunction(NextcloudBackend.rename)
def test_move_is_async(self):
"""move must be an async coroutine on NextcloudBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.nextcloud_backend import NextcloudBackend
assert hasattr(NextcloudBackend, "move"), (
"NextcloudBackend must implement move (Phase 13)"
)
assert inspect.iscoroutinefunction(NextcloudBackend.move)
def test_delete_is_async(self):
"""delete must be an async coroutine on NextcloudBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.nextcloud_backend import NextcloudBackend
assert hasattr(NextcloudBackend, "delete"), (
"NextcloudBackend must implement delete (Phase 13)"
)
assert inspect.iscoroutinefunction(NextcloudBackend.delete)
def test_upload_file_is_async(self):
"""upload_file must be an async coroutine on NextcloudBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.nextcloud_backend import NextcloudBackend
assert hasattr(NextcloudBackend, "upload_file"), (
"NextcloudBackend must implement upload_file (Phase 13)"
)
assert inspect.iscoroutinefunction(NextcloudBackend.upload_file)
@pytest.mark.asyncio
async def test_create_folder_ssrf_guarded(self):
"""create_folder must validate target path against SSRF redirect attempts (T-13-04).
Nextcloud WebDAV create-folder (MKCOL) responses can include a Location header
pointing to an internal address. The backend must validate that any redirect
destination is within the configured server_url host.
FAILS: create_folder SSRF guard does not exist yet.
"""
from storage.nextcloud_backend import NextcloudBackend
from storage.cloud_utils import validate_cloud_url
backend = self._make_backend()
# The backend must reject any MKCOL path that tries to escape the server_url
backend._client.mkdir.side_effect = Exception("Internal redirect to http://169.254.169.254/")
result = await backend.create_folder(
parent_ref=None,
name="../../etc",
connection_id=_uuid.uuid4(),
user_id=_uuid.uuid4(),
)
assert isinstance(result, dict), "create_folder must return normalized dict even on error"
assert result.get("kind") in ("folder", "conflict", "error"), (
f"Unexpected kind: {result.get('kind')!r}"
)
@pytest.mark.asyncio
async def test_delete_normalizes_result(self):
"""NextcloudBackend delete returns normalized {'kind': 'deleted', 'reason': 'permanent'}.
Nextcloud WebDAV does not expose a standard Trash API — deletes are permanent
unless the Trash plugin is active and specifically targeted. Default = permanent.
FAILS: delete method does not exist yet.
"""
from storage.nextcloud_backend import NextcloudBackend
backend = self._make_backend()
backend._client.clean.return_value = True
result = await backend.delete("remote.php/dav/files/user/report.pdf")
assert isinstance(result, dict), "delete must return normalized dict"
assert result.get("kind") == "deleted"
assert result.get("reason") in ("permanent", "trashed"), (
f"Expected 'permanent' or 'trashed', got {result.get('reason')!r}"
)
class TestWebDAVMutableContract:
"""WebDAV — Phase 13 mutable-operation contract (RED).
T-13-04: WebDAV outbound SSRF guard must be preserved during mutations.
All method assertions will fail until Phase 13 implements the mutable interface.
"""
def _make_backend(self):
from storage.webdav_backend import WebDAVBackend
with patch("storage.webdav_backend.validate_cloud_url"):
with patch("webdav3.client.Client"):
backend = WebDAVBackend.__new__(WebDAVBackend)
backend._server_url = "https://dav.example.com/dav/"
backend._client = MagicMock()
return backend
def test_create_folder_is_async(self):
"""create_folder must be an async coroutine on WebDAVBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.webdav_backend import WebDAVBackend
assert hasattr(WebDAVBackend, "create_folder"), (
"WebDAVBackend must implement create_folder (Phase 13)"
)
assert inspect.iscoroutinefunction(WebDAVBackend.create_folder)
def test_rename_is_async(self):
"""rename must be an async coroutine on WebDAVBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.webdav_backend import WebDAVBackend
assert hasattr(WebDAVBackend, "rename"), (
"WebDAVBackend must implement rename (Phase 13)"
)
assert inspect.iscoroutinefunction(WebDAVBackend.rename)
def test_move_is_async(self):
"""move must be an async coroutine on WebDAVBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.webdav_backend import WebDAVBackend
assert hasattr(WebDAVBackend, "move"), (
"WebDAVBackend must implement move (Phase 13)"
)
assert inspect.iscoroutinefunction(WebDAVBackend.move)
def test_delete_is_async(self):
"""delete must be an async coroutine on WebDAVBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.webdav_backend import WebDAVBackend
assert hasattr(WebDAVBackend, "delete"), (
"WebDAVBackend must implement delete (Phase 13)"
)
assert inspect.iscoroutinefunction(WebDAVBackend.delete)
def test_upload_file_is_async(self):
"""upload_file must be an async coroutine on WebDAVBackend (Phase 13).
FAILS: method does not exist yet.
"""
import inspect
from storage.webdav_backend import WebDAVBackend
assert hasattr(WebDAVBackend, "upload_file"), (
"WebDAVBackend must implement upload_file (Phase 13)"
)
assert inspect.iscoroutinefunction(WebDAVBackend.upload_file)
@pytest.mark.asyncio
async def test_delete_permanent_normalizes_result(self):
"""WebDAVBackend delete returns normalized {'kind': 'deleted', 'reason': 'permanent'}.
D-11: Generic WebDAV has no trash API — deletes are permanent.
The confirmation dialog must say so explicitly.
FAILS: delete method does not exist yet.
"""
from storage.webdav_backend import WebDAVBackend
backend = self._make_backend()
backend._client.clean.return_value = True
result = await backend.delete("/dav/report.pdf")
assert isinstance(result, dict), "delete must return normalized dict"
assert result.get("kind") == "deleted"
assert result.get("reason") == "permanent", (
"WebDAV deletes are permanent — reason must be 'permanent' (D-11)"
)
@pytest.mark.asyncio
async def test_move_validates_destination_stays_on_same_host(self):
"""WebDAV MOVE destination must be validated to stay on the configured server host.
T-13-04: A WebDAV MOVE request accepts a Destination header. The backend must
ensure the destination URL cannot escape to a different host (SSRF via Destination).
FAILS: move SSRF guard does not exist yet.
"""
from storage.webdav_backend import WebDAVBackend
backend = self._make_backend()
# Attempt to move to a destination on a different host
try:
result = await backend.move(
provider_item_id="report.pdf",
destination_parent_ref="http://169.254.169.254/evil/",
etag="v1",
)
# If it returns (not raises), it must produce a typed error result
assert isinstance(result, dict)
assert result.get("kind") == "invalid_destination", (
f"Expected kind='invalid_destination' for SSRF destination, got {result.get('kind')!r}"
)
except (ValueError, Exception) as exc:
# Raising an exception with the SSRF block is also acceptable
assert "SSRF" in str(exc) or "host" in str(exc).lower() or "invalid" in str(exc).lower(), (
f"SSRF move rejection must mention 'SSRF', 'host', or 'invalid', got: {exc}"
)
def test_no_direct_cloud_items_writes(self):
"""WebDAVBackend mutation methods must not import or write cloud_items directly.
Provider-neutral contract: adapters must never write to cloud_items. Only the
service layer (services.cloud_items) is allowed to reconcile metadata.
FAILS: Phase 13 implementation not yet done — this test will pass once
we verify the import graph doesn't include cloud_items in the backend module.
"""
import importlib
import sys
# Load the module source to check imports
spec = importlib.util.find_spec("storage.webdav_backend")
if spec and spec.origin:
with open(spec.origin) as f:
source = f.read()
assert "from services.cloud_items" not in source and "import cloud_items" not in source, (
"WebDAVBackend must not import from services.cloud_items — "
"only the service layer writes cloud metadata (provider-neutral contract)"
)