kite/backend/tests/test_cloud_provider_contract.py

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


# =============================================================================
# Phase 13 Plan 01 — RED: Four-provider mutable-operation parity contract
#
# All tests below FAIL until Phase 13 adds MutableCloudResourceAdapter.
# =============================================================================

# ── Phase 13 mutable provider cases ──────────────────────────────────────────

MUTABLE_PROVIDER_CASES = [
    pytest.param("nextcloud", _make_nextcloud_adapter, id="nextcloud"),
    pytest.param("webdav", _make_webdav_adapter, id="webdav"),
    pytest.param("google_drive", _make_google_drive_adapter, id="google_drive"),
    pytest.param("onedrive", _make_onedrive_adapter, id="onedrive"),
]

# ── Mutable method signatures — canonical contract ────────────────────────────

MUTABLE_METHODS = [
    "create_folder",
    "rename",
    "move",
    "delete",
    "upload_file",
]


class TestMutableAdapterContract:
    """
    Phase 13: Four-provider mutable-operation parity.

    Each provider must implement the same mutable-adapter method signatures.
    The canonical contract is provider-neutral:
      - No direct cloud_items writes in any adapter.
      - No browse-time byte transfer.
      - No hidden overwrite path.
      - Result types use normalized kind/reason dicts.
      - Caller identity (connection_id, user_id) flows through every method.

    All tests FAIL until Phase 13 adds the mutable adapter contract.
    """

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_mutable_methods_exist(self, provider, factory_fn):
        """All Phase 13 mutable methods must exist on every provider adapter.

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        for method_name in MUTABLE_METHODS:
            assert hasattr(adapter, method_name), (
                f"{provider}: missing Phase 13 method '{method_name}' — "
                "all four providers must implement the mutable adapter contract"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_mutable_methods_are_async(self, provider, factory_fn):
        """All Phase 13 mutable methods must be async coroutines.

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        for method_name in MUTABLE_METHODS:
            method = getattr(type(adapter), method_name, None)
            if method is None:
                pytest.fail(f"{provider}: {method_name} not found — implement Phase 13 contract")
            assert inspect.iscoroutinefunction(method), (
                f"{provider}: {method_name} must be defined with 'async def'"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_create_folder_signature(self, provider, factory_fn):
        """create_folder must accept (parent_ref, name, connection_id, user_id).

        Provider-neutral contract: caller identity must be available to the method
        for audit trail and reconciliation handoff. No provider-specific extra kwargs.

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        assert hasattr(adapter, "create_folder"), (
            f"{provider}: create_folder not found"
        )
        sig = inspect.signature(adapter.create_folder)
        params = list(sig.parameters.keys())
        for required in ("parent_ref", "name", "connection_id", "user_id"):
            assert required in params, (
                f"{provider}: create_folder missing required param '{required}'; "
                f"got params: {params}"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_rename_signature(self, provider, factory_fn):
        """rename must accept (provider_item_id, new_name, etag).

        etag is required for stale-metadata detection (D-07).

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        assert hasattr(adapter, "rename"), f"{provider}: rename not found"
        sig = inspect.signature(adapter.rename)
        params = list(sig.parameters.keys())
        for required in ("provider_item_id", "new_name", "etag"):
            assert required in params, (
                f"{provider}: rename missing required param '{required}'; got: {params}"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_move_signature(self, provider, factory_fn):
        """move must accept (provider_item_id, destination_parent_ref, etag).

        etag ensures moves are not applied to stale metadata (D-07, D-08).

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        assert hasattr(adapter, "move"), f"{provider}: move not found"
        sig = inspect.signature(adapter.move)
        params = list(sig.parameters.keys())
        for required in ("provider_item_id", "destination_parent_ref", "etag"):
            assert required in params, (
                f"{provider}: move missing required param '{required}'; got: {params}"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_delete_signature(self, provider, factory_fn):
        """delete must accept (provider_item_id) with optional trash kwarg.

        D-11: Provider trash/recycle-bin preference must be expressible by the caller.
        The default should be trash=True (prefer trash when supported).

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        assert hasattr(adapter, "delete"), f"{provider}: delete not found"
        sig = inspect.signature(adapter.delete)
        params = list(sig.parameters.keys())
        assert "provider_item_id" in params, (
            f"{provider}: delete missing 'provider_item_id'; got: {params}"
        )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_upload_file_signature(self, provider, factory_fn):
        """upload_file must accept (parent_ref, filename, content, content_type, connection_id, user_id).

        connection_id and user_id are required for audit trail and reconciliation.

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        assert hasattr(adapter, "upload_file"), f"{provider}: upload_file not found"
        sig = inspect.signature(adapter.upload_file)
        params = list(sig.parameters.keys())
        for required in ("parent_ref", "filename", "connection_id", "user_id"):
            assert required in params, (
                f"{provider}: upload_file missing required param '{required}'; got: {params}"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_no_direct_cloud_items_imports(self, provider, factory_fn):
        """Provider adapter modules must not import from services.cloud_items.

        Provider-neutral contract: adapters must never write cloud metadata directly.
        Only the service layer is authorised to call reconcile_cloud_listing or
        upsert_cloud_item.

        FAILS until Phase 13 implementation confirms the import boundary.
        """
        import importlib
        import sys

        # Map provider names to module paths
        module_map = {
            "nextcloud": "storage.nextcloud_backend",
            "webdav": "storage.webdav_backend",
            "google_drive": "storage.google_drive_backend",
            "onedrive": "storage.onedrive_backend",
        }
        module_name = module_map.get(provider)
        if not module_name:
            pytest.skip(f"No module map for provider {provider!r}")

        spec = importlib.util.find_spec(module_name)
        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, (
                f"{provider}: adapter module must not import from services.cloud_items — "
                "only the service layer writes cloud metadata (provider-neutral contract)"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_no_browse_time_byte_transfer(self, provider, factory_fn):
        """list_folder must not call upload_file, delete, rename, move, or create_folder.

        Provider-neutral contract: browse operations are read-only. Mutable methods
        must never be invoked as a side effect of listing.

        This test passes on the existing list_folder implementations and must
        remain green through Phase 13 implementation.
        """
        adapter = factory_fn()
        for mutable_method in MUTABLE_METHODS:
            if hasattr(adapter, mutable_method):
                method = getattr(adapter, mutable_method)
                # Spy setup would need actual invocation; check this is a separate method
                # (not aliased to list_folder)
                assert method is not getattr(adapter, "list_folder", None), (
                    f"{provider}: {mutable_method} must not be aliased to list_folder"
                )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_unsupported_capability_is_disclosed(self, provider, factory_fn):
        """get_capabilities must explicitly disclose which mutation actions are unsupported.

        Provider-neutral contract: structurally unsupported actions must appear
        in the capabilities dict with state='unsupported' and a known reason code.
        The adapter must not silently omit unsupported actions from capabilities.

        This test passes on the existing get_capabilities contract and must remain
        green through Phase 13 Phase implementation.
        """
        from storage.cloud_base import (
            CloudCapability, ACTIONS, STATE_SUPPORTED,
            STATE_UNSUPPORTED, STATE_TEMPORARILY_UNAVAILABLE,
        )

        # Verify ACTIONS includes all Phase 13 mutation actions
        expected_mutation_actions = {"create_folder", "rename", "move", "delete", "upload"}
        for action in expected_mutation_actions:
            assert action in ACTIONS, (
                f"cloud_base.ACTIONS must include mutation action '{action}' for Phase 13"
            )


class TestMutableAdapterResultNormalization:
    """
    Phase 13 Plan 01 — RED: Normalized result type assertions for mutable operations.

    Every mutation method must return a dict with at minimum {'kind': ..., 'reason': ...}.
    Callers must never parse raw provider error structures — normalization happens
    inside the adapter, not in the router or service layer.
    """

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_normalized_result_type_contract(self, provider, factory_fn):
        """Mutable operations must be documented to return normalized dicts.

        This test asserts the existence of mutable methods and that their docstrings
        declare the return type, giving Phase 13 implementors a clear failing signal.

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        for method_name in MUTABLE_METHODS:
            assert hasattr(adapter, method_name), (
                f"{provider}: {method_name} must be implemented for Phase 13"
            )
            method = getattr(adapter, method_name)
            # Method must have a docstring (normalized result docs are required)
            assert method.__doc__, (
                f"{provider}.{method_name} must have a docstring describing the normalized return type"
            )

    @pytest.mark.parametrize("provider,factory_fn", MUTABLE_PROVIDER_CASES)
    def test_conflict_normalization_defined(self, provider, factory_fn):
        """The adapter must have a way to normalize provider-specific conflict errors.

        Provider-neutral contract: conflict normalization must happen inside the adapter.
        No router or service layer should pattern-match against raw provider error classes.

        FAILS: Phase 13 adapter methods not yet implemented.
        """
        adapter = factory_fn()
        # The adapter must expose a normalization helper or internal method for conflicts.
        # Acceptable patterns: _normalize_error, _handle_conflict, _map_error, etc.
        normalization_methods = [
            "_normalize_error",
            "_handle_conflict",
            "_map_error",
            "_classify_error",
            "_normalize_result",
        ]
        has_normalization = any(hasattr(adapter, m) for m in normalization_methods)
        # Also acceptable: if the mutable methods themselves handle normalization via
        # a documented try/except that returns typed dicts. This assertion gives a
        # failing signal that prompts the implementor to add normalization.
        assert has_normalization or any(
            hasattr(adapter, m) for m in MUTABLE_METHODS
        ), (
            f"{provider}: adapter must implement Phase 13 mutable methods with "
            f"normalized conflict/error handling (no raw provider exceptions in response)"
        )