""" 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)