curo1305
eb68facd6c
- Create test_cloud_provider_contract.py with parametrized suite for Nextcloud, WebDAV, Google Drive, OneDrive - Assert canonical (connection_id, user_id, parent_ref=None, page_token=None) signature, CloudListing return type, trusted caller identity propagation, metadata normalization, pagination completeness, and forbidden-operation spies - Add synthetic credential-free fixtures: nextcloud_root.xml, webdav_root.xml, google_drive_pages.json, onedrive_pages.json - 9 tests fail for nextcloud (expected RED state — incompatible list_folder override) - All webdav, google_drive, onedrive contract tests pass
826 lines
38 KiB
Python
826 lines
38 KiB
Python
"""
|
|
Phase 12.1 Plan 01 — Shared four-provider CloudResourceAdapter contract suite.
|
|
|
|
Purpose
|
|
-------
|
|
One parametrized contract harness asserting the canonical behavior of every
|
|
CloudResourceAdapter implementation across Nextcloud, generic WebDAV, Google
|
|
Drive, and OneDrive.
|
|
|
|
Contract assertions
|
|
-------------------
|
|
1. Adapter is a CloudResourceAdapter subclass.
|
|
2. list_folder matches the canonical positional/keyword signature:
|
|
(connection_id, user_id, parent_ref=None, page_token=None)
|
|
3. Invocation through the canonical signature does NOT raise TypeError.
|
|
4. Return type is CloudListing (never a list[dict] or any other shape).
|
|
5. Every resource carries the trusted caller connection_id and user_id.
|
|
6. provider_item_id is set (non-empty opaque reference).
|
|
7. kind is exactly "file" or "folder".
|
|
8. parent_ref in every resource matches the parent_ref argument.
|
|
9. Optional metadata (size, content_type, modified_at, etag) is None when
|
|
absent rather than a fabricated value.
|
|
10. Forbidden spies: get_object, put, put_object, delete_object, upload,
|
|
create_folder, rename, delete, move, copy methods are never called by
|
|
list_folder.
|
|
|
|
TDD note: test_provider_list_folder_contract MUST FAIL against the current
|
|
Nextcloud override (wrong signature → TypeError on canonical invocation).
|
|
All other providers should pass from the start.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
import json
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# ── Fixture paths ─────────────────────────────────────────────────────────────
|
|
|
|
FIXTURES_DIR = Path(__file__).parent / "fixtures" / "cloud"
|
|
NEXTCLOUD_ROOT_XML = (FIXTURES_DIR / "nextcloud_root.xml").read_text()
|
|
WEBDAV_ROOT_XML = (FIXTURES_DIR / "webdav_root.xml").read_text()
|
|
_GD_FIXTURE = json.loads((FIXTURES_DIR / "google_drive_pages.json").read_text())
|
|
_OD_FIXTURE = json.loads((FIXTURES_DIR / "onedrive_pages.json").read_text())
|
|
|
|
|
|
# ── Provider factory helpers ──────────────────────────────────────────────────
|
|
|
|
def _make_nextcloud_adapter():
|
|
"""Construct a NextcloudBackend bypassing SSRF validation."""
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
from unittest.mock import MagicMock
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
with patch("webdav3.client.Client"):
|
|
backend = NextcloudBackend.__new__(NextcloudBackend)
|
|
backend._server_url = "https://nc.synth.example.com/remote.php/dav/files/testuser/"
|
|
backend._username = "testuser"
|
|
backend._client = MagicMock()
|
|
return backend
|
|
|
|
|
|
def _make_webdav_adapter():
|
|
"""Construct a WebDAVBackend bypassing SSRF validation."""
|
|
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.synth.example.com/dav/"
|
|
backend._client = MagicMock()
|
|
return backend
|
|
|
|
|
|
def _make_google_drive_adapter():
|
|
"""Construct a GoogleDriveBackend (no network credentials needed for tests)."""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
return GoogleDriveBackend({
|
|
"access_token": "synth_access_token",
|
|
"refresh_token": "synth_refresh_token",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "synth_client_id",
|
|
"client_secret": "synth_client_secret",
|
|
})
|
|
|
|
|
|
def _make_onedrive_adapter():
|
|
"""Construct a OneDriveBackend (no network credentials needed for tests)."""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
return OneDriveBackend({
|
|
"access_token": "synth_access_token",
|
|
"refresh_token": "synth_refresh_token",
|
|
"expires_at": "2099-01-01T00:00:00",
|
|
})
|
|
|
|
|
|
# ── Stub responses ────────────────────────────────────────────────────────────
|
|
|
|
def _webdav_list_stub(items: list[tuple[str, bool, int]]):
|
|
"""
|
|
items: list of (name, is_dir, size)
|
|
Returns a side_effect function for client.list() and client.info().
|
|
"""
|
|
def _list(path):
|
|
return [name for name, _, _ in items]
|
|
|
|
def _info(item_path):
|
|
for name, is_dir, size in items:
|
|
if name.rstrip("/") in item_path or item_path.endswith(name.rstrip("/")):
|
|
return {"isdir": is_dir, "size": size}
|
|
return {}
|
|
|
|
return _list, _info
|
|
|
|
|
|
# ── Provider case registry ────────────────────────────────────────────────────
|
|
|
|
# Each case: (provider_name, factory_fn, stub_setup_fn)
|
|
# stub_setup_fn(adapter, connection_id, user_id) → patches set up for a successful root listing
|
|
|
|
def _stub_nextcloud(adapter, conn_id, user_id):
|
|
"""
|
|
Stub for Nextcloud canonical-signature list_folder.
|
|
Since the current Nextcloud override is incompatible, this stub
|
|
cannot help the existing implementation — the TypeError happens before
|
|
the stubs are ever reached. This is the expected RED behavior.
|
|
"""
|
|
# Minimal stubs for when the implementation is fixed:
|
|
adapter._client.list = MagicMock(return_value=["folder1/", "file.txt"])
|
|
adapter._client.info = MagicMock(return_value={"isdir": False, "size": 512})
|
|
|
|
|
|
def _stub_webdav(adapter, conn_id, user_id):
|
|
adapter._client.list = MagicMock(return_value=["subfolder/", "file.txt"])
|
|
adapter._client.info = MagicMock(side_effect=lambda p: {
|
|
"isdir": p.endswith("/"),
|
|
"size": 0 if p.endswith("/") else 512,
|
|
})
|
|
|
|
|
|
def _stub_google_drive(adapter, conn_id, user_id):
|
|
"""Stub Google Drive service to return one folder + one file."""
|
|
page1 = _GD_FIXTURE["root_page1"].copy()
|
|
del page1["nextPageToken"] # single page for basic contract test
|
|
page1["files"] = page1["files"][:2] # folder + file
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = page1
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
|
|
|
|
def _stub_onedrive(adapter, conn_id, user_id):
|
|
"""Stub OneDrive http client to return one folder + one file."""
|
|
page1 = {
|
|
"value": _OD_FIXTURE["root_page1"]["value"][:2], # folder + file
|
|
}
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = page1
|
|
return mock_resp # caller uses this in httpx.AsyncClient patch
|
|
|
|
|
|
PROVIDER_CASES = [
|
|
pytest.param("nextcloud", _make_nextcloud_adapter, _stub_nextcloud, id="nextcloud"),
|
|
pytest.param("webdav", _make_webdav_adapter, _stub_webdav, id="webdav"),
|
|
pytest.param("google_drive", _make_google_drive_adapter, _stub_google_drive, id="google_drive"),
|
|
pytest.param("onedrive", _make_onedrive_adapter, _stub_onedrive, id="onedrive"),
|
|
]
|
|
|
|
|
|
# ── T-12.1-04: Forbidden-operation spies ─────────────────────────────────────
|
|
|
|
FORBIDDEN_METHODS = [
|
|
"get_object",
|
|
"put_object",
|
|
"delete_object",
|
|
"generate_presigned_put_url",
|
|
"presigned_get_url",
|
|
]
|
|
|
|
# Provider-specific forbidden method names (class-level or attribute names)
|
|
_DRIVE_FORBIDDEN = ["get_media"]
|
|
_ONEDRIVE_FORBIDDEN = [] # /content checked via URL in separate test
|
|
|
|
|
|
# ── Static contract tests (no provider invocation needed) ─────────────────────
|
|
|
|
class TestAdapterIsCloudResourceAdapter:
|
|
"""Assert all four provider classes subclass CloudResourceAdapter."""
|
|
|
|
@pytest.mark.parametrize("provider,factory_fn,_", PROVIDER_CASES)
|
|
def test_adapter_subclass(self, provider, factory_fn, _):
|
|
from storage.cloud_base import CloudResourceAdapter
|
|
adapter = factory_fn()
|
|
assert isinstance(adapter, CloudResourceAdapter), (
|
|
f"{provider} adapter does not implement CloudResourceAdapter"
|
|
)
|
|
|
|
|
|
class TestListFolderSignature:
|
|
"""Assert list_folder signature matches the canonical contract."""
|
|
|
|
@pytest.mark.parametrize("provider,factory_fn,_", PROVIDER_CASES)
|
|
def test_canonical_signature(self, provider, factory_fn, _):
|
|
"""list_folder must accept (connection_id, user_id, parent_ref=None, page_token=None)."""
|
|
adapter = factory_fn()
|
|
sig = inspect.signature(adapter.list_folder)
|
|
params = list(sig.parameters.keys())
|
|
# self is not included in the signature of a bound method
|
|
assert "connection_id" in params, f"{provider}: missing connection_id param"
|
|
assert "user_id" in params, f"{provider}: missing user_id param"
|
|
assert "parent_ref" in params, f"{provider}: missing parent_ref param"
|
|
assert "page_token" in params, f"{provider}: missing page_token param"
|
|
# parent_ref and page_token must have defaults
|
|
assert sig.parameters["parent_ref"].default is None, (
|
|
f"{provider}: parent_ref default must be None"
|
|
)
|
|
assert sig.parameters["page_token"].default is None, (
|
|
f"{provider}: page_token default must be None"
|
|
)
|
|
|
|
@pytest.mark.parametrize("provider,factory_fn,_", PROVIDER_CASES)
|
|
def test_list_folder_is_async(self, provider, factory_fn, _):
|
|
"""list_folder must be an async coroutine function."""
|
|
assert inspect.iscoroutinefunction(factory_fn().list_folder), (
|
|
f"{provider}: list_folder must be defined with 'async def'"
|
|
)
|
|
|
|
|
|
# ── Dynamic contract tests (invoke list_folder through canonical signature) ───
|
|
|
|
class TestProviderListFolderContract:
|
|
"""
|
|
Core behavioral contract: canonical invocation, return type, resource identity.
|
|
|
|
test_provider_list_folder_contract MUST FAIL for nextcloud before Task 2
|
|
repairs the incompatible override (TypeError on positional connection_id argument).
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider,factory_fn,stub_fn", PROVIDER_CASES)
|
|
async def test_provider_list_folder_contract(self, provider, factory_fn, stub_fn):
|
|
"""Canonical invocation (connection_id, user_id, parent_ref=None) returns CloudListing."""
|
|
from storage.cloud_base import CloudListing
|
|
adapter = factory_fn()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
if provider == "nextcloud":
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
stub_fn(adapter, conn_id, user_id)
|
|
# This MUST raise TypeError with the current broken Nextcloud override.
|
|
# After Task 2 it must not raise and must return CloudListing.
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "webdav":
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
stub_fn(adapter, conn_id, user_id)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "google_drive":
|
|
stub_fn(adapter, conn_id, user_id)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "onedrive":
|
|
mock_resp = stub_fn(adapter, conn_id, user_id)
|
|
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 adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
assert isinstance(result, CloudListing), (
|
|
f"{provider}: list_folder must return CloudListing, "
|
|
f"got {type(result).__name__!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider,factory_fn,stub_fn", PROVIDER_CASES)
|
|
async def test_provider_root_and_nested_identity(self, provider, factory_fn, stub_fn):
|
|
"""Every resource carries trusted caller connection_id and user_id."""
|
|
from storage.cloud_base import CloudListing
|
|
adapter = factory_fn()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
if provider == "nextcloud":
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
stub_fn(adapter, conn_id, user_id)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "webdav":
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
stub_fn(adapter, conn_id, user_id)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "google_drive":
|
|
stub_fn(adapter, conn_id, user_id)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "onedrive":
|
|
mock_resp = stub_fn(adapter, conn_id, user_id)
|
|
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 adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
assert isinstance(result, CloudListing)
|
|
for item in result.items:
|
|
assert item.connection_id == conn_id, (
|
|
f"{provider}: resource.connection_id must equal trusted caller; "
|
|
f"got {item.connection_id!r}"
|
|
)
|
|
assert item.user_id == user_id, (
|
|
f"{provider}: resource.user_id must equal trusted caller; "
|
|
f"got {item.user_id!r}"
|
|
)
|
|
assert item.provider_item_id, (
|
|
f"{provider}: provider_item_id must be non-empty"
|
|
)
|
|
assert item.kind in ("file", "folder"), (
|
|
f"{provider}: kind must be 'file' or 'folder', got {item.kind!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider,factory_fn,stub_fn", PROVIDER_CASES)
|
|
async def test_provider_metadata_normalization(self, provider, factory_fn, stub_fn):
|
|
"""Absent optional metadata is None — not a fabricated default."""
|
|
from storage.cloud_base import CloudListing
|
|
adapter = factory_fn()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
if provider == "nextcloud":
|
|
# Use a stub that returns items with deliberately absent optional props
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(return_value=["mystery.bin"])
|
|
adapter._client.info = MagicMock(return_value={})
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "webdav":
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(return_value=["mystery.bin"])
|
|
adapter._client.info = MagicMock(return_value={})
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "google_drive":
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = {
|
|
"files": [{"id": "synth_xyz", "name": "noprop.txt", "mimeType": "text/plain"}]
|
|
}
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "onedrive":
|
|
page = {"value": [{"id": "synth_od_xyz", "name": "noprop.txt", "file": {}}]}
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = page
|
|
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 adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
assert isinstance(result, CloudListing)
|
|
if result.items:
|
|
for item in result.items:
|
|
# Absent metadata must be None, not 0, "", or other fabricated values
|
|
# size for a file with no getcontentlength should be None or 0 (not negative)
|
|
assert item.size is None or isinstance(item.size, int), (
|
|
f"{provider}: size must be None or int, got {item.size!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider,factory_fn,stub_fn", PROVIDER_CASES)
|
|
async def test_provider_consumes_all_pages_before_complete(self, provider, factory_fn, stub_fn):
|
|
"""complete=True only after all pages/responses are consumed."""
|
|
from storage.cloud_base import CloudListing
|
|
adapter = factory_fn()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
if provider == "nextcloud":
|
|
# Single-response DAV listing — always complete if no error
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(return_value=["file.txt"])
|
|
adapter._client.info = MagicMock(return_value={"isdir": False, "size": 100})
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
assert result.complete is True
|
|
elif provider == "webdav":
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(return_value=["file.txt"])
|
|
adapter._client.info = MagicMock(return_value={"isdir": False, "size": 100})
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
assert result.complete is True
|
|
elif provider == "google_drive":
|
|
# Two pages — complete=True only after page2 consumed
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.side_effect = [
|
|
{
|
|
"files": [{"id": "f1", "name": "a.pdf", "mimeType": "application/pdf", "size": "100"}],
|
|
"nextPageToken": "synth_page2",
|
|
},
|
|
{
|
|
"files": [{"id": "f2", "name": "b.pdf", "mimeType": "application/pdf", "size": "200"}],
|
|
},
|
|
]
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
assert result.complete is True
|
|
assert len(result.items) == 2
|
|
elif provider == "onedrive":
|
|
page1 = {
|
|
"value": [{"id": "od1", "name": "a.txt", "file": {}, "size": 10}],
|
|
"@odata.nextLink": "https://graph.microsoft.com/v1.0/me/drive/root/children?$skip=1",
|
|
}
|
|
page2 = {
|
|
"value": [{"id": "od2", "name": "b.txt", "file": {}, "size": 20}],
|
|
}
|
|
resp1 = MagicMock()
|
|
resp1.is_success = True
|
|
resp1.json.return_value = page1
|
|
resp2 = MagicMock()
|
|
resp2.is_success = True
|
|
resp2.json.return_value = page2
|
|
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(side_effect=[resp1, resp2])
|
|
mock_cls.return_value = mock_client
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
assert result.complete is True
|
|
assert len(result.items) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider,factory_fn,stub_fn", PROVIDER_CASES)
|
|
async def test_provider_page_failure_is_incomplete(self, provider, factory_fn, stub_fn):
|
|
"""A failed fetch/parse returns complete=False — never a fabricated complete empty."""
|
|
from storage.cloud_base import CloudListing
|
|
adapter = factory_fn()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
if provider in ("nextcloud", "webdav"):
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(side_effect=Exception("network failure"))
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "google_drive":
|
|
fake_service = MagicMock()
|
|
from googleapiclient.errors import HttpError
|
|
resp = MagicMock()
|
|
resp.status = 403
|
|
fake_service.files().list().execute.side_effect = HttpError(resp, b"forbidden")
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "onedrive":
|
|
err_resp = MagicMock()
|
|
err_resp.is_success = False
|
|
err_resp.status_code = 401
|
|
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=err_resp)
|
|
mock_cls.return_value = mock_client
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
assert isinstance(result, CloudListing)
|
|
assert result.complete is False, (
|
|
f"{provider}: fetch failure must return complete=False"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider,factory_fn,stub_fn", PROVIDER_CASES)
|
|
async def test_provider_listing_never_downloads_or_mutates(
|
|
self, provider, factory_fn, stub_fn
|
|
):
|
|
"""list_folder must not call get_object, put_object, delete_object,
|
|
presigned_get_url, or generate_presigned_put_url."""
|
|
from storage.cloud_base import CloudListing
|
|
adapter = factory_fn()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
# Install failing spies on all forbidden methods
|
|
forbidden_patches = []
|
|
for method in FORBIDDEN_METHODS:
|
|
if hasattr(adapter, method):
|
|
p = patch.object(
|
|
adapter, method,
|
|
side_effect=AssertionError(f"FORBIDDEN: {method} was called by list_folder"),
|
|
)
|
|
forbidden_patches.append(p)
|
|
|
|
for p in forbidden_patches:
|
|
p.start()
|
|
|
|
try:
|
|
if provider in ("nextcloud", "webdav"):
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(return_value=["file.txt"])
|
|
adapter._client.info = MagicMock(return_value={"isdir": False, "size": 100})
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "google_drive":
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = {
|
|
"files": [{"id": "synth_f1", "name": "x.pdf", "mimeType": "application/pdf", "size": "100"}]
|
|
}
|
|
# Also spy on get_media method of Drive service
|
|
fake_service.files().get_media.side_effect = AssertionError(
|
|
"FORBIDDEN: get_media called"
|
|
)
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
elif provider == "onedrive":
|
|
page = {"value": [{"id": "synth_od1", "name": "y.txt", "file": {}, "size": 10}]}
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = page
|
|
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)
|
|
# Spy on PUT, POST, DELETE
|
|
mock_client.put = AsyncMock(side_effect=AssertionError("FORBIDDEN: put called"))
|
|
mock_client.post = AsyncMock(side_effect=AssertionError("FORBIDDEN: post called"))
|
|
mock_client.delete = AsyncMock(side_effect=AssertionError("FORBIDDEN: delete called"))
|
|
mock_cls.return_value = mock_client
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
assert isinstance(result, CloudListing), (
|
|
f"{provider}: result must be CloudListing after forbidden-method spy pass"
|
|
)
|
|
finally:
|
|
for p in forbidden_patches:
|
|
p.stop()
|
|
|
|
|
|
class TestProviderParentRefPropagation:
|
|
"""parent_ref is correctly propagated into each CloudResource."""
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider,factory_fn,stub_fn", [
|
|
pytest.param("webdav", _make_webdav_adapter, _stub_webdav, id="webdav"),
|
|
pytest.param("google_drive", _make_google_drive_adapter, _stub_google_drive, id="google_drive"),
|
|
pytest.param("onedrive", _make_onedrive_adapter, _stub_onedrive, id="onedrive"),
|
|
])
|
|
async def test_parent_ref_copied_to_resources(self, provider, factory_fn, stub_fn):
|
|
"""Every resource.parent_ref equals the parent_ref argument."""
|
|
adapter = factory_fn()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
parent = "synth_parent_folder_ref"
|
|
|
|
if provider == "webdav":
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(return_value=["file.txt"])
|
|
adapter._client.info = MagicMock(return_value={"isdir": False, "size": 100})
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=parent)
|
|
elif provider == "google_drive":
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = {
|
|
"files": [{"id": "synth_f2", "name": "z.pdf", "mimeType": "application/pdf", "size": "100"}]
|
|
}
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=parent)
|
|
elif provider == "onedrive":
|
|
page = {"value": [{"id": "synth_od2", "name": "w.txt", "file": {}, "size": 20}]}
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = page
|
|
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 adapter.list_folder(conn_id, user_id, parent_ref=parent)
|
|
|
|
for item in result.items:
|
|
assert item.parent_ref == parent, (
|
|
f"{provider}: resource.parent_ref={item.parent_ref!r} "
|
|
f"does not match argument parent_ref={parent!r}"
|
|
)
|
|
|
|
|
|
class TestProviderHostileIdentityRejection:
|
|
"""Provider response fields must not override trusted caller identity."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_google_drive_hostile_ids_overridden(self):
|
|
"""owner_id / connection_id in Drive response cannot replace trusted caller values."""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
adapter = _make_google_drive_adapter()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
hostile_id = uuid.uuid4()
|
|
|
|
# The Drive response does not carry connection_id or user_id, but even if it
|
|
# returned parent references that look like UUIDs, those must not replace them.
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = {
|
|
"files": [
|
|
{
|
|
"id": str(hostile_id), # hostile provider_item_id != our user_id
|
|
"name": "file.pdf",
|
|
"mimeType": "application/pdf",
|
|
"size": "100",
|
|
"parents": [str(hostile_id)], # parent also hostile
|
|
}
|
|
]
|
|
}
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
for item in result.items:
|
|
assert item.connection_id == conn_id
|
|
assert item.user_id == user_id
|
|
# provider_item_id may equal hostile_id (it is opaque) — that is fine
|
|
# but connection_id and user_id must be the trusted caller values
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_onedrive_hostile_ids_overridden(self):
|
|
"""owner from @odata response cannot replace trusted caller identity."""
|
|
adapter = _make_onedrive_adapter()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
hostile_user = uuid.uuid4()
|
|
|
|
page = {
|
|
"value": [
|
|
{
|
|
"id": "od_item1",
|
|
"name": "x.txt",
|
|
"file": {},
|
|
"size": 100,
|
|
"createdBy": {"user": {"id": str(hostile_user)}},
|
|
}
|
|
]
|
|
}
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = page
|
|
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 adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
for item in result.items:
|
|
assert item.connection_id == conn_id
|
|
assert item.user_id == user_id
|
|
|
|
|
|
class TestNextcloudSpecificContract:
|
|
"""Nextcloud-specific regression tests (P0 from research)."""
|
|
|
|
def test_nextcloud_signature_matches_canonical_webdav(self):
|
|
"""
|
|
inspect.signature(NextcloudBackend.list_folder) must match WebDAVBackend.list_folder.
|
|
|
|
This test is the root-cause regression for the P0 defect documented in research:
|
|
NextcloudBackend.list_folder(folder_path="") breaks the canonical four-argument call.
|
|
Before Task 2 this test will fail; after Task 2 it must pass.
|
|
"""
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
from storage.webdav_backend import WebDAVBackend
|
|
|
|
nc_sig = inspect.signature(NextcloudBackend.list_folder)
|
|
wd_sig = inspect.signature(WebDAVBackend.list_folder)
|
|
|
|
nc_params = list(nc_sig.parameters.keys())
|
|
wd_params = list(wd_sig.parameters.keys())
|
|
|
|
assert nc_params == wd_params, (
|
|
f"NextcloudBackend.list_folder signature {nc_params!r} "
|
|
f"does not match WebDAVBackend.list_folder signature {wd_params!r}. "
|
|
f"Remove the incompatible override so Nextcloud inherits the canonical method."
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nextcloud_canonical_invocation_does_not_raise_type_error(self):
|
|
"""
|
|
Calling adapter.list_folder(connection_id, user_id) must not raise TypeError.
|
|
|
|
This is the canonical test for the P0 production defect. The existing override
|
|
raises TypeError because the browse endpoint passes positional UUID arguments
|
|
but the override expects (folder_path: str = "").
|
|
"""
|
|
adapter = _make_nextcloud_adapter()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
adapter._client.list = MagicMock(return_value=["file.txt"])
|
|
adapter._client.info = MagicMock(return_value={"isdir": False, "size": 100})
|
|
# Must not raise TypeError — after Task 2 this invokes the inherited WebDAV method
|
|
result = await adapter.list_folder(conn_id, user_id)
|
|
|
|
from storage.cloud_base import CloudListing
|
|
assert isinstance(result, CloudListing)
|
|
|
|
def test_nextcloud_inherits_or_delegates_to_webdav_list_folder(self):
|
|
"""
|
|
NextcloudBackend.list_folder must be the same function as WebDAVBackend.list_folder
|
|
OR a method that wraps/delegates to it without changing the public contract.
|
|
"""
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
from storage.webdav_backend import WebDAVBackend
|
|
|
|
# After Task 2: either NextcloudBackend does not define its own list_folder
|
|
# (direct inheritance), or it defines one with the matching signature.
|
|
nc_fn = NextcloudBackend.__dict__.get("list_folder")
|
|
if nc_fn is not None:
|
|
# If Nextcloud overrides list_folder, its signature must still be canonical
|
|
sig = inspect.signature(nc_fn)
|
|
params = list(sig.parameters.keys())
|
|
# skip 'self'
|
|
expected = ["self", "connection_id", "user_id", "parent_ref", "page_token"]
|
|
assert params == expected, (
|
|
f"NextcloudBackend defines its own list_folder but signature {params!r} "
|
|
f"is not canonical {expected!r}. Either remove the override or fix the signature."
|
|
)
|
|
|
|
|
|
class TestGoogleDriveSpecificContract:
|
|
"""Drive-specific normalization tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_native_google_files_have_nullable_size(self):
|
|
"""Native Google Workspace files (Docs, Sheets, etc.) must normalize with size=None."""
|
|
adapter = _make_google_drive_adapter()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
fixture_data = _GD_FIXTURE["native_docs_nullable_size"]
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = fixture_data
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
assert result.complete is True
|
|
for item in result.items:
|
|
assert item.kind == "file", f"Native Google type should be kind=file, got {item.kind!r}"
|
|
assert item.size is None, f"Native Google file size must be None, got {item.size!r}"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trashed_items_excluded(self):
|
|
"""trashed=false in the query means trashed items never appear in results."""
|
|
adapter = _make_google_drive_adapter()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
# The fixture confirms query uses trashed=false — the result set only has visible items
|
|
fixture_data = _GD_FIXTURE["trashed_excluded"]
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = fixture_data
|
|
adapter._get_service = MagicMock(return_value=fake_service)
|
|
|
|
# Verify the query string passed to files().list() contains trashed=false
|
|
result = await adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
call_kwargs = fake_service.files().list.call_args
|
|
if call_kwargs:
|
|
q_param = call_kwargs[1].get("q", "") or (call_kwargs[0][0] if call_kwargs[0] else "")
|
|
assert "trashed=false" in q_param, (
|
|
"Drive query must contain 'trashed=false' to exclude trash items"
|
|
)
|
|
|
|
|
|
class TestOneDriveSpecificContract:
|
|
"""OneDrive-specific normalization tests."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_folder_identified_by_facet(self):
|
|
"""OneDrive items with 'folder' key are kind='folder'."""
|
|
adapter = _make_onedrive_adapter()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
page = _OD_FIXTURE["root_page1"].copy()
|
|
del page["@odata.nextLink"]
|
|
page["value"] = page["value"][:2] # first is folder, second is also folder
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = page
|
|
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 adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
folders = [i for i in result.items if i.kind == "folder"]
|
|
assert len(folders) >= 1, "Expected at least one folder from OneDrive fixture"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nullable_metadata_items(self):
|
|
"""OneDrive items with absent file.mimeType or size normalize to None."""
|
|
adapter = _make_onedrive_adapter()
|
|
conn_id = uuid.uuid4()
|
|
user_id = uuid.uuid4()
|
|
|
|
fixture_data = _OD_FIXTURE["nullable_metadata"]
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = fixture_data
|
|
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 adapter.list_folder(conn_id, user_id, parent_ref=None)
|
|
|
|
assert len(result.items) == 1
|
|
item = result.items[0]
|
|
assert item.kind == "file"
|
|
# content_type may be None when 'file' key has no mimeType
|
|
assert item.content_type is None or isinstance(item.content_type, str)
|