curo1305
f47e36d93d
- TestUploadConflictSemantics: typed upload results across all 4 providers - TestKeepBothNaming: keep_both_name() helper with counter before extension - Tests verify credentials never appear in upload results (T-13-17) - Upload kind values must be MUT_KINDS constants - 16/23 pass (providers already have upload_file); 7 fail (keep_both_name missing)
1997 lines
82 KiB
Python
1997 lines
82 KiB
Python
"""
|
|
Phase 5 Plan 03 — TDD RED tests for GoogleDriveBackend and OneDriveBackend.
|
|
|
|
These tests are written BEFORE the implementation files exist, to serve as the
|
|
RED phase of the TDD cycle. They verify:
|
|
|
|
1. GoogleDriveBackend:
|
|
- Importability of class and CloudConnectionError
|
|
- All 7 methods are async coroutines
|
|
- presigned_get_url raises NotImplementedError
|
|
- generate_presigned_put_url raises NotImplementedError
|
|
- CloudConnectionError has a reason attribute
|
|
- cache_discovery=False is used (prevents /tmp traversal — T-05-03-05)
|
|
|
|
2. OneDriveBackend:
|
|
- Importability of class
|
|
- All 7 methods are async coroutines
|
|
- CHUNK_SIZE = 10 * 1024 * 1024 (10 MB — Pitfall 6 prevention)
|
|
- presigned_get_url raises NotImplementedError
|
|
- generate_presigned_put_url raises NotImplementedError
|
|
- CloudConnectionError imported from google_drive_backend (shared exception type)
|
|
- _ensure_valid_token skips refresh when token is not expired (non-expired expires_at)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import uuid as _uuid
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
# ── GoogleDriveBackend tests ──────────────────────────────────────────────────
|
|
|
|
class TestGoogleDriveBackendImport:
|
|
"""Verify the module and class can be imported."""
|
|
|
|
def test_google_drive_backend_import(self):
|
|
"""GoogleDriveBackend and CloudConnectionError are importable."""
|
|
from storage.google_drive_backend import GoogleDriveBackend, CloudConnectionError # noqa: F401
|
|
assert GoogleDriveBackend is not None
|
|
assert CloudConnectionError is not None
|
|
|
|
def test_cloud_connection_error_has_reason(self):
|
|
"""CloudConnectionError stores a reason attribute."""
|
|
from storage.google_drive_backend import CloudConnectionError
|
|
err = CloudConnectionError("test", reason="token_expired")
|
|
assert err.reason == "token_expired"
|
|
|
|
def test_cloud_connection_error_reason_invalid_grant(self):
|
|
"""CloudConnectionError stores reason='invalid_grant'."""
|
|
from storage.google_drive_backend import CloudConnectionError
|
|
err = CloudConnectionError("", reason="invalid_grant")
|
|
assert err.reason == "invalid_grant"
|
|
|
|
|
|
class TestGoogleDriveBackendMethods:
|
|
"""Verify all 7 StorageBackend methods are implemented as async coroutines."""
|
|
|
|
METHODS = [
|
|
"put_object",
|
|
"get_object",
|
|
"delete_object",
|
|
"presigned_get_url",
|
|
"health_check",
|
|
"generate_presigned_put_url",
|
|
"stat_object",
|
|
]
|
|
|
|
def _make_backend(self):
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
return GoogleDriveBackend({
|
|
"access_token": "fake_token",
|
|
"refresh_token": "fake_refresh",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "fake_client_id",
|
|
"client_secret": "fake_client_secret",
|
|
})
|
|
|
|
@pytest.mark.parametrize("method_name", METHODS)
|
|
def test_method_is_coroutine(self, method_name):
|
|
"""Every StorageBackend method must be an async coroutine function."""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
method = getattr(GoogleDriveBackend, method_name)
|
|
assert inspect.iscoroutinefunction(method), (
|
|
f"{method_name} must be defined with 'async def'"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_presigned_get_url_raises_not_implemented(self):
|
|
"""presigned_get_url raises NotImplementedError (D-14 — cloud backends use get_object)."""
|
|
backend = self._make_backend()
|
|
with pytest.raises(NotImplementedError):
|
|
await backend.presigned_get_url("some_file_id")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_presigned_put_url_raises_not_implemented(self):
|
|
"""generate_presigned_put_url raises NotImplementedError (D-14)."""
|
|
backend = self._make_backend()
|
|
with pytest.raises(NotImplementedError):
|
|
await backend.generate_presigned_put_url("some_file_id")
|
|
|
|
def test_inherits_storage_backend(self):
|
|
"""GoogleDriveBackend is a StorageBackend subclass."""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
from storage.base import StorageBackend
|
|
assert issubclass(GoogleDriveBackend, StorageBackend)
|
|
|
|
|
|
class TestGoogleDriveBackendInit:
|
|
"""Verify __init__ constructs a valid Credentials object from a dict."""
|
|
|
|
def test_init_with_valid_credentials_dict(self):
|
|
"""GoogleDriveBackend.__init__ accepts the credentials dict and creates _creds."""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
creds_dict = {
|
|
"access_token": "ya29.test",
|
|
"refresh_token": "1//test_refresh",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "test_client_id",
|
|
"client_secret": "test_client_secret",
|
|
"expiry": "2099-01-01T00:00:00",
|
|
}
|
|
backend = GoogleDriveBackend(creds_dict)
|
|
# Verify credentials object was created
|
|
assert backend._creds is not None
|
|
assert backend._creds.token == "ya29.test"
|
|
assert backend._creds.refresh_token == "1//test_refresh"
|
|
|
|
def test_init_stores_creds_dict(self):
|
|
"""GoogleDriveBackend stores the original credentials dict."""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
creds_dict = {
|
|
"access_token": "ya29.test",
|
|
"refresh_token": "1//test_refresh",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "test_client_id",
|
|
"client_secret": "test_client_secret",
|
|
}
|
|
backend = GoogleDriveBackend(creds_dict)
|
|
assert backend._creds_dict is creds_dict
|
|
|
|
|
|
# ── OneDriveBackend tests ─────────────────────────────────────────────────────
|
|
|
|
class TestOneDriveBackendImport:
|
|
"""Verify OneDriveBackend is importable and uses shared CloudConnectionError."""
|
|
|
|
def test_onedrive_backend_import(self):
|
|
"""OneDriveBackend is importable from storage.onedrive_backend."""
|
|
from storage.onedrive_backend import OneDriveBackend # noqa: F401
|
|
assert OneDriveBackend is not None
|
|
|
|
def test_chunk_size(self):
|
|
"""CHUNK_SIZE must be exactly 10 MB (Pitfall 6 — above Graph 4 MB limit)."""
|
|
from storage.onedrive_backend import CHUNK_SIZE
|
|
assert CHUNK_SIZE == 10 * 1024 * 1024, (
|
|
f"CHUNK_SIZE should be 10 * 1024 * 1024 (10 MB), got {CHUNK_SIZE}"
|
|
)
|
|
|
|
def test_cloud_connection_error_is_shared(self):
|
|
"""CloudConnectionError used in OneDriveBackend is the same class from google_drive_backend."""
|
|
from storage.google_drive_backend import CloudConnectionError as GDriveError
|
|
from storage.onedrive_backend import CloudConnectionError as OneDriveError
|
|
assert GDriveError is OneDriveError, (
|
|
"OneDriveBackend must import CloudConnectionError from google_drive_backend, "
|
|
"not define its own."
|
|
)
|
|
|
|
|
|
class TestOneDriveBackendMethods:
|
|
"""Verify all 7 StorageBackend methods are async on OneDriveBackend."""
|
|
|
|
METHODS = [
|
|
"put_object",
|
|
"get_object",
|
|
"delete_object",
|
|
"presigned_get_url",
|
|
"health_check",
|
|
"generate_presigned_put_url",
|
|
"stat_object",
|
|
]
|
|
|
|
def _make_backend(self, expires_at: str = "2099-01-01T00:00:00"):
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
return OneDriveBackend({
|
|
"access_token": "fake_access",
|
|
"refresh_token": "fake_refresh",
|
|
"expires_at": expires_at,
|
|
})
|
|
|
|
@pytest.mark.parametrize("method_name", METHODS)
|
|
def test_method_is_coroutine(self, method_name):
|
|
"""Every StorageBackend method must be async."""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
method = getattr(OneDriveBackend, method_name)
|
|
assert inspect.iscoroutinefunction(method), (
|
|
f"{method_name} must be defined with 'async def'"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_presigned_get_url_raises_not_implemented(self):
|
|
"""presigned_get_url raises NotImplementedError (D-14)."""
|
|
backend = self._make_backend()
|
|
with pytest.raises(NotImplementedError):
|
|
await backend.presigned_get_url("item_id")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generate_presigned_put_url_raises_not_implemented(self):
|
|
"""generate_presigned_put_url raises NotImplementedError (D-14)."""
|
|
backend = self._make_backend()
|
|
with pytest.raises(NotImplementedError):
|
|
await backend.generate_presigned_put_url("item_id")
|
|
|
|
def test_inherits_storage_backend(self):
|
|
"""OneDriveBackend is a StorageBackend subclass."""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
from storage.base import StorageBackend
|
|
assert issubclass(OneDriveBackend, StorageBackend)
|
|
|
|
|
|
class TestOneDriveBackendInit:
|
|
"""Verify OneDriveBackend __init__ stores credentials correctly."""
|
|
|
|
def test_init_stores_credentials(self):
|
|
"""OneDriveBackend stores the credentials dict as _credentials."""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
creds = {
|
|
"access_token": "Bearer test",
|
|
"refresh_token": "test_refresh",
|
|
"expires_at": "2099-01-01T00:00:00",
|
|
}
|
|
backend = OneDriveBackend(creds)
|
|
assert backend._credentials is creds
|
|
|
|
def test_auth_headers_contain_bearer(self):
|
|
"""_auth_headers() returns Authorization: Bearer <access_token>."""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
creds = {
|
|
"access_token": "my_token_123",
|
|
"refresh_token": "test_refresh",
|
|
"expires_at": "2099-01-01T00:00:00",
|
|
}
|
|
backend = OneDriveBackend(creds)
|
|
headers = backend._auth_headers()
|
|
assert headers["Authorization"] == "Bearer my_token_123"
|
|
|
|
|
|
class TestOneDriveEnsureValidToken:
|
|
"""Test _ensure_valid_token logic for expired / non-expired tokens."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_token_no_refresh_needed(self):
|
|
"""Non-expired token should not trigger a refresh."""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
backend = OneDriveBackend({
|
|
"access_token": "valid_token",
|
|
"refresh_token": "refresh",
|
|
"expires_at": "2099-01-01T00:00:00",
|
|
})
|
|
# Should complete without error (no msal call needed)
|
|
await backend._ensure_valid_token()
|
|
# Token should remain unchanged
|
|
assert backend._credentials["access_token"] == "valid_token"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_expired_token_raises_cloud_connection_error_on_invalid_grant(self):
|
|
"""Expired token with invalid_grant from MSAL raises CloudConnectionError(reason='invalid_grant')."""
|
|
from unittest.mock import patch, AsyncMock, MagicMock
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
from storage.google_drive_backend import CloudConnectionError
|
|
|
|
backend = OneDriveBackend({
|
|
"access_token": "expired_token",
|
|
"refresh_token": "bad_refresh",
|
|
"expires_at": "2000-01-01T00:00:00", # clearly expired
|
|
})
|
|
|
|
# Mock _refresh_token to return None (simulating invalid_grant)
|
|
async def fake_refresh():
|
|
return None
|
|
|
|
with patch.object(backend, "_refresh_token", side_effect=fake_refresh):
|
|
with pytest.raises(CloudConnectionError) as exc_info:
|
|
await backend._ensure_valid_token()
|
|
assert exc_info.value.reason == "invalid_grant"
|
|
|
|
|
|
# ── Phase 12: CloudResourceAdapter contract tests ────────────────────────────
|
|
|
|
class TestCloudResourceAdapterContract:
|
|
"""Verify all four providers implement the CloudResourceAdapter interface."""
|
|
|
|
def test_google_drive_is_adapter(self):
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
from storage.cloud_base import CloudResourceAdapter
|
|
assert issubclass(GoogleDriveBackend, CloudResourceAdapter)
|
|
|
|
def test_onedrive_is_adapter(self):
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
from storage.cloud_base import CloudResourceAdapter
|
|
assert issubclass(OneDriveBackend, CloudResourceAdapter)
|
|
|
|
def test_webdav_is_adapter(self):
|
|
from storage.webdav_backend import WebDAVBackend
|
|
from storage.cloud_base import CloudResourceAdapter
|
|
assert issubclass(WebDAVBackend, CloudResourceAdapter)
|
|
|
|
def test_nextcloud_is_adapter(self):
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
from storage.cloud_base import CloudResourceAdapter
|
|
assert issubclass(NextcloudBackend, CloudResourceAdapter)
|
|
|
|
def test_factory_build_cloud_resource_adapter(self):
|
|
from storage.cloud_backend_factory import build_cloud_resource_adapter
|
|
from storage.cloud_base import CloudResourceAdapter
|
|
adapter = build_cloud_resource_adapter(
|
|
"google_drive",
|
|
{
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "cid",
|
|
"client_secret": "csec",
|
|
},
|
|
)
|
|
assert isinstance(adapter, CloudResourceAdapter)
|
|
|
|
def test_factory_rejects_unknown_provider(self):
|
|
from storage.cloud_backend_factory import build_cloud_resource_adapter
|
|
with pytest.raises(ValueError, match="Unknown provider"):
|
|
build_cloud_resource_adapter("s3", {})
|
|
|
|
|
|
class TestGoogleDriveListFolder:
|
|
"""list_folder must never call get_object, put, delete, or move."""
|
|
|
|
def _make_backend(self):
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
return GoogleDriveBackend({
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "cid",
|
|
"client_secret": "csec",
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_returns_cloud_listing(self):
|
|
from storage.cloud_base import CloudListing
|
|
from unittest.mock import MagicMock, patch
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = {
|
|
"files": [
|
|
{"id": "f1", "name": "Report.pdf", "mimeType": "application/pdf", "size": "1024"},
|
|
{"id": "d1", "name": "Folder", "mimeType": "application/vnd.google-apps.folder"},
|
|
]
|
|
}
|
|
with patch.object(backend, "_get_service", return_value=fake_service):
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
assert isinstance(result, CloudListing)
|
|
assert len(result.items) == 2
|
|
file_item = next(i for i in result.items if i.kind == "file")
|
|
folder_item = next(i for i in result.items if i.kind == "folder")
|
|
assert file_item.name == "Report.pdf"
|
|
assert folder_item.name == "Folder"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_follows_pagination(self):
|
|
from storage.cloud_base import CloudListing
|
|
from unittest.mock import MagicMock, patch, call
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
# First page returns nextPageToken; second page does not
|
|
fake_service.files().list().execute.side_effect = [
|
|
{
|
|
"files": [{"id": "f1", "name": "File1.pdf", "mimeType": "application/pdf", "size": "100"}],
|
|
"nextPageToken": "tok2",
|
|
},
|
|
{
|
|
"files": [{"id": "f2", "name": "File2.pdf", "mimeType": "application/pdf", "size": "200"}],
|
|
},
|
|
]
|
|
with patch.object(backend, "_get_service", return_value=fake_service):
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
assert result.complete is True
|
|
assert len(result.items) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_native_doc_size_is_none(self):
|
|
from unittest.mock import MagicMock, patch
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = {
|
|
"files": [
|
|
{
|
|
"id": "gdoc1",
|
|
"name": "My Doc",
|
|
"mimeType": "application/vnd.google-apps.document",
|
|
},
|
|
]
|
|
}
|
|
with patch.object(backend, "_get_service", return_value=fake_service):
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
assert len(result.items) == 1
|
|
assert result.items[0].size is None
|
|
assert result.items[0].kind == "file" # native docs are not folders
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_never_calls_get_object(self):
|
|
"""list_folder must never invoke get_object (byte download guard)."""
|
|
from unittest.mock import MagicMock, patch
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
fake_service.files().list().execute.return_value = {"files": []}
|
|
with patch.object(backend, "_get_service", return_value=fake_service):
|
|
with patch.object(backend, "get_object", side_effect=AssertionError("get_object called")) as mock_get:
|
|
await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
mock_get.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_incomplete_on_http_error(self):
|
|
from unittest.mock import MagicMock, patch
|
|
from googleapiclient.errors import HttpError
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status = 403
|
|
fake_service.files().list().execute.side_effect = HttpError(resp, b"forbidden")
|
|
with patch.object(backend, "_get_service", return_value=fake_service):
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
assert result.complete is False
|
|
|
|
|
|
class TestOneDriveListFolder:
|
|
"""list_folder must never call get_object, put, delete, or move."""
|
|
|
|
def _make_backend(self):
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
return OneDriveBackend({
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"expires_at": "2099-01-01T00:00:00",
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_returns_cloud_listing(self):
|
|
from storage.cloud_base import CloudListing
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
import uuid
|
|
import httpx
|
|
|
|
backend = self._make_backend()
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = {
|
|
"value": [
|
|
{"id": "item1", "name": "Doc.docx", "file": {"mimeType": "application/vnd.openxmlformats-officedocument.wordprocessingml.document"}, "size": 4096},
|
|
{"id": "folder1", "name": "Projects", "folder": {"childCount": 3}},
|
|
]
|
|
}
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_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_client_cls.return_value = mock_client
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
assert isinstance(result, CloudListing)
|
|
assert len(result.items) == 2
|
|
file_item = next(i for i in result.items if i.kind == "file")
|
|
folder_item = next(i for i in result.items if i.kind == "folder")
|
|
assert file_item.name == "Doc.docx"
|
|
assert folder_item.name == "Projects"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_never_calls_get_object(self):
|
|
"""list_folder must never invoke get_object."""
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = {"value": []}
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_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_client_cls.return_value = mock_client
|
|
with patch.object(backend, "get_object", side_effect=AssertionError("get_object called")) as mock_get:
|
|
await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
mock_get.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_follows_odata_next_link(self):
|
|
"""list_folder follows @odata.nextLink until exhausted."""
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
page1 = MagicMock()
|
|
page1.is_success = True
|
|
page1.json.return_value = {
|
|
"value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}],
|
|
"@odata.nextLink": "https://graph.microsoft.com/v1.0/me/drive/root/children?$skip=1",
|
|
}
|
|
page2 = MagicMock()
|
|
page2.is_success = True
|
|
page2.json.return_value = {"value": [{"id": "i2", "name": "B.txt", "file": {}, "size": 20}]}
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_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=[page1, page2])
|
|
mock_client_cls.return_value = mock_client
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
assert result.complete is True
|
|
assert len(result.items) == 2
|
|
|
|
|
|
class TestOneDriveCrossOriginNextLink:
|
|
"""OneDrive @odata.nextLink must stay on graph.microsoft.com — T-12.1-03."""
|
|
|
|
def _make_backend(self):
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
return OneDriveBackend({
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"expires_at": "2099-01-01T00:00:00",
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_origin_next_link_returns_incomplete(self):
|
|
"""A cross-origin @odata.nextLink must stop pagination and return complete=False."""
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
page1 = MagicMock()
|
|
page1.is_success = True
|
|
page1.json.return_value = {
|
|
"value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}],
|
|
# Cross-origin nextLink — must be rejected
|
|
"@odata.nextLink": "https://evil.example.com/v1.0/me/drive/root/children?$skip=1",
|
|
}
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_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=page1)
|
|
mock_client_cls.return_value = mock_client
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
# Items from page1 should be retained
|
|
assert len(result.items) == 1
|
|
assert result.items[0].name == "A.txt"
|
|
# But listing is not complete (cross-origin continuation not trusted)
|
|
assert result.complete is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_origin_graph_next_link_followed(self):
|
|
"""A same-origin graph.microsoft.com @odata.nextLink is trusted and followed."""
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
page1 = MagicMock()
|
|
page1.is_success = True
|
|
page1.json.return_value = {
|
|
"value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}],
|
|
"@odata.nextLink": "https://graph.microsoft.com/v1.0/me/drive/root/children?$skip=1",
|
|
}
|
|
page2 = MagicMock()
|
|
page2.is_success = True
|
|
page2.json.return_value = {
|
|
"value": [{"id": "i2", "name": "B.txt", "file": {}, "size": 20}],
|
|
}
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_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=[page1, page2])
|
|
mock_client_cls.return_value = mock_client
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
assert result.complete is True
|
|
assert len(result.items) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_page_failure_retains_prior_items(self):
|
|
"""A page HTTP failure returns complete=False but retains already-normalized items."""
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
page1 = MagicMock()
|
|
page1.is_success = True
|
|
page1.json.return_value = {
|
|
"value": [{"id": "i1", "name": "A.txt", "file": {}, "size": 10}],
|
|
"@odata.nextLink": "https://graph.microsoft.com/v1.0/me/drive/root/children?$skip=1",
|
|
}
|
|
page2_fail = MagicMock()
|
|
page2_fail.is_success = False
|
|
page2_fail.status_code = 503
|
|
|
|
with patch("httpx.AsyncClient") as mock_client_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=[page1, page2_fail])
|
|
mock_client_cls.return_value = mock_client
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
# Page1 items retained
|
|
assert len(result.items) == 1
|
|
assert result.items[0].name == "A.txt"
|
|
# Not complete because page2 failed
|
|
assert result.complete is False
|
|
|
|
|
|
class TestGoogleDriveAuthFailureControl:
|
|
"""Drive auth failures map to controlled complete=False — T-12.1-03."""
|
|
|
|
def _make_backend(self):
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
return GoogleDriveBackend({
|
|
"access_token": "tok",
|
|
"refresh_token": "ref",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "cid",
|
|
"client_secret": "csec",
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_401_returns_incomplete_not_raises(self):
|
|
"""401 from Drive returns complete=False — not an uncaught exception."""
|
|
from unittest.mock import MagicMock, patch
|
|
from googleapiclient.errors import HttpError
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
resp = MagicMock()
|
|
resp.status = 401
|
|
fake_service.files().list().execute.side_effect = HttpError(resp, b"unauthorized")
|
|
with patch.object(backend, "_get_service", return_value=fake_service):
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
assert result.complete is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_page1_ok_page2_error_retains_items(self):
|
|
"""Error on page2 still returns items from page1 with complete=False."""
|
|
from unittest.mock import MagicMock, patch
|
|
from googleapiclient.errors import HttpError
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
resp403 = MagicMock()
|
|
resp403.status = 403
|
|
fake_service.files().list().execute.side_effect = [
|
|
{
|
|
"files": [{"id": "f1", "name": "a.pdf", "mimeType": "application/pdf", "size": "100"}],
|
|
"nextPageToken": "synth_tok",
|
|
},
|
|
HttpError(resp403, b"scope"),
|
|
]
|
|
with patch.object(backend, "_get_service", return_value=fake_service):
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
# Items from page1 must be retained
|
|
assert len(result.items) == 1
|
|
assert result.items[0].name == "a.pdf"
|
|
assert result.complete is False
|
|
|
|
|
|
class TestWebDAVListFolder:
|
|
"""WebDAV list_folder tests — SSRF guard and no byte-download."""
|
|
|
|
def _make_backend(self):
|
|
from storage.webdav_backend import WebDAVBackend
|
|
from unittest.mock import MagicMock, patch
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
with patch("webdav3.client.Client"):
|
|
backend = WebDAVBackend.__new__(WebDAVBackend)
|
|
backend._server_url = "https://dav.example.com/"
|
|
backend._client = MagicMock()
|
|
return backend
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_returns_cloud_listing(self):
|
|
from storage.cloud_base import CloudListing
|
|
from unittest.mock import patch, MagicMock
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
|
|
def fake_list(path):
|
|
return ["file.txt", "subdir/"]
|
|
|
|
def fake_info(path):
|
|
if "subdir" in path:
|
|
return {"isdir": True, "size": 0}
|
|
return {"isdir": False, "size": 512, "getcontenttype": "text/plain"}
|
|
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
backend._client.list.side_effect = fake_list
|
|
backend._client.info.side_effect = fake_info
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
assert isinstance(result, CloudListing)
|
|
assert len(result.items) == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_never_calls_get_object(self):
|
|
"""list_folder must never invoke get_object."""
|
|
from unittest.mock import patch, MagicMock
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
backend._client.list.return_value = []
|
|
with patch.object(backend, "get_object", side_effect=AssertionError("get_object called")) as mock_get:
|
|
await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
mock_get.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_incomplete_on_error(self):
|
|
from unittest.mock import patch
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
backend._client.list.side_effect = Exception("connection failed")
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
assert result.complete is False
|
|
assert len(result.items) == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_folder_absent_props_treated_as_unknown(self):
|
|
"""Items with missing PROPFIND fields are included with None values, not excluded."""
|
|
from unittest.mock import patch
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
|
|
def fake_list(path):
|
|
return ["mystery.bin"]
|
|
|
|
def fake_info(path):
|
|
return {}
|
|
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
backend._client.list.side_effect = fake_list
|
|
backend._client.info.side_effect = fake_info
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
assert len(result.items) == 1
|
|
item = result.items[0]
|
|
assert item.name == "mystery.bin"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_capabilities_returns_all_actions(self):
|
|
from storage.cloud_base import ACTIONS
|
|
from unittest.mock import patch
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
backend._client.check.return_value = True
|
|
caps = await backend.get_capabilities(uuid.uuid4(), uuid.uuid4())
|
|
|
|
assert set(caps.keys()) == ACTIONS
|
|
assert caps["browse"].state == "supported"
|
|
|
|
|
|
# =============================================================================
|
|
# Phase 13 Plan 01 — RED: Provider-specific mutable-operation contracts
|
|
#
|
|
# All tests below FAIL until Phase 13 implements the mutable adapter interface.
|
|
# =============================================================================
|
|
|
|
|
|
class TestGoogleDriveMutableContract:
|
|
"""Google Drive — Phase 13 mutable-operation contract (RED).
|
|
|
|
D-17: Phase 13 requests broader drive.files scope — test must assert
|
|
the consent scope includes 'drive' or 'drive.file' as the minimum for
|
|
listing and operating on pre-existing Drive items.
|
|
|
|
All method assertions will fail until MutableCloudResourceAdapter is
|
|
implemented in Phase 13.
|
|
"""
|
|
|
|
def _make_backend(self):
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
return GoogleDriveBackend({
|
|
"access_token": "ya29.test_13",
|
|
"refresh_token": "1//test_refresh_13",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "test_client_13",
|
|
"client_secret": "test_secret_13",
|
|
})
|
|
|
|
def test_create_folder_is_async(self):
|
|
"""create_folder must be an async coroutine on GoogleDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
assert hasattr(GoogleDriveBackend, "create_folder"), (
|
|
"GoogleDriveBackend must implement create_folder (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(GoogleDriveBackend.create_folder), (
|
|
"create_folder must be async"
|
|
)
|
|
|
|
def test_rename_is_async(self):
|
|
"""rename must be an async coroutine on GoogleDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
assert hasattr(GoogleDriveBackend, "rename"), (
|
|
"GoogleDriveBackend must implement rename (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(GoogleDriveBackend.rename)
|
|
|
|
def test_move_is_async(self):
|
|
"""move must be an async coroutine on GoogleDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
assert hasattr(GoogleDriveBackend, "move"), (
|
|
"GoogleDriveBackend must implement move (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(GoogleDriveBackend.move)
|
|
|
|
def test_delete_is_async(self):
|
|
"""delete must be an async coroutine on GoogleDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
assert hasattr(GoogleDriveBackend, "delete"), (
|
|
"GoogleDriveBackend must implement delete (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(GoogleDriveBackend.delete)
|
|
|
|
def test_upload_file_is_async(self):
|
|
"""upload_file must be an async coroutine on GoogleDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
assert hasattr(GoogleDriveBackend, "upload_file"), (
|
|
"GoogleDriveBackend must implement upload_file (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(GoogleDriveBackend.upload_file)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_prefers_trash_normalizes_result(self):
|
|
"""Google Drive delete should prefer Trash and return typed result dict.
|
|
|
|
D-11: Provider trash is preferred over permanent delete. The normalized
|
|
result must carry {'kind': 'deleted', 'reason': 'trashed'} or 'permanent'.
|
|
|
|
FAILS: delete method does not exist yet.
|
|
"""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
fake_service.files().trash().execute.return_value = {}
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
|
|
result = await backend.delete("file_id_123", trash=True)
|
|
assert isinstance(result, dict), "delete must return a normalized dict result"
|
|
assert result.get("kind") == "deleted", (
|
|
f"Expected kind='deleted', got {result.get('kind')!r}"
|
|
)
|
|
assert result.get("reason") in ("trashed", "permanent"), (
|
|
f"Expected reason 'trashed' or 'permanent', got {result.get('reason')!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_folder_name_collision_normalizes_result(self):
|
|
"""create_folder with a collision returns normalized conflict kind/reason.
|
|
|
|
D-05: Create-folder collision must return {'kind': 'conflict', 'reason': 'name_collision'}
|
|
rather than raising an unhandled API error.
|
|
|
|
FAILS: create_folder method does not exist yet.
|
|
"""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
# Simulate Drive returning a 409 or duplicate error
|
|
from googleapiclient.errors import HttpError
|
|
fake_resp = MagicMock()
|
|
fake_resp.status = 409
|
|
fake_service.files().create().execute.side_effect = HttpError(
|
|
resp=fake_resp, content=b'{"error":{"message":"File already exists"}}'
|
|
)
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
|
|
result = await backend.create_folder(
|
|
parent_ref=None,
|
|
name="Existing Folder",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
assert isinstance(result, dict), "create_folder must return a normalized dict"
|
|
assert result.get("kind") == "conflict", (
|
|
f"Expected kind='conflict', got {result.get('kind')!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rename_stale_metadata_normalizes_result(self):
|
|
"""rename with mismatched etag returns normalized stale kind/reason.
|
|
|
|
D-07: Stale rename must return {'kind': 'stale', 'reason': 'item_changed'}.
|
|
|
|
FAILS: rename method does not exist yet.
|
|
"""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
|
|
backend = self._make_backend()
|
|
fake_service = MagicMock()
|
|
from googleapiclient.errors import HttpError
|
|
fake_resp = MagicMock()
|
|
fake_resp.status = 412
|
|
fake_service.files().update().execute.side_effect = HttpError(
|
|
resp=fake_resp, content=b'{"error":{"message":"precondition failed"}}'
|
|
)
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
|
|
result = await backend.rename(
|
|
provider_item_id="file_id_123",
|
|
new_name="Renamed.pdf",
|
|
etag="old-etag",
|
|
)
|
|
assert isinstance(result, dict), "rename must return a normalized dict"
|
|
assert result.get("kind") == "stale", (
|
|
f"Expected kind='stale', got {result.get('kind')!r}"
|
|
)
|
|
|
|
def test_drive_scope_includes_full_drive_access(self):
|
|
"""Google Drive backend must request 'drive' or 'drive.file' scope (D-17).
|
|
|
|
D-17: Phase 13 requests broader Google Drive access — 'drive.file' scope only
|
|
allows access to files created by the app. Full drive browsing requires 'drive'.
|
|
|
|
FAILS: Phase 13 scope expansion not yet implemented.
|
|
"""
|
|
import inspect
|
|
import re
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
|
|
# The backend's OAuth scope configuration must include full drive access
|
|
# Check for SCOPES constant or _SCOPES attribute
|
|
scopes = getattr(GoogleDriveBackend, "SCOPES", None) or \
|
|
getattr(GoogleDriveBackend, "_SCOPES", None) or \
|
|
getattr(GoogleDriveBackend, "OAUTH_SCOPES", None)
|
|
|
|
assert scopes is not None, (
|
|
"GoogleDriveBackend must declare a SCOPES / _SCOPES / OAUTH_SCOPES class attribute "
|
|
"for the OAuth consent scope (D-17)"
|
|
)
|
|
scope_str = " ".join(scopes) if isinstance(scopes, (list, tuple)) else str(scopes)
|
|
# Phase 13 requires at minimum 'drive' scope (not just 'drive.file' or 'drive.readonly')
|
|
assert "https://www.googleapis.com/auth/drive" in scope_str or "drive" in scope_str, (
|
|
f"GoogleDriveBackend SCOPES must include full Drive access for Phase 13 (D-17), "
|
|
f"got: {scope_str!r}"
|
|
)
|
|
|
|
|
|
class TestOneDriveMutableContract:
|
|
"""OneDrive — Phase 13 mutable-operation contract (RED).
|
|
|
|
CONN-02: Successful token refresh must hand the new access_token and
|
|
refresh_token off to the service layer for encrypted persistence.
|
|
|
|
All method assertions will fail until Phase 13 implements the mutable interface.
|
|
"""
|
|
|
|
def _make_backend(self, expires_at: str = "2099-01-01T00:00:00"):
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
return OneDriveBackend({
|
|
"access_token": "od_access_13",
|
|
"refresh_token": "od_refresh_13",
|
|
"expires_at": expires_at,
|
|
})
|
|
|
|
def test_create_folder_is_async(self):
|
|
"""create_folder must be an async coroutine on OneDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
assert hasattr(OneDriveBackend, "create_folder"), (
|
|
"OneDriveBackend must implement create_folder (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(OneDriveBackend.create_folder)
|
|
|
|
def test_rename_is_async(self):
|
|
"""rename must be an async coroutine on OneDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
assert hasattr(OneDriveBackend, "rename"), (
|
|
"OneDriveBackend must implement rename (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(OneDriveBackend.rename)
|
|
|
|
def test_move_is_async(self):
|
|
"""move must be an async coroutine on OneDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
assert hasattr(OneDriveBackend, "move"), (
|
|
"OneDriveBackend must implement move (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(OneDriveBackend.move)
|
|
|
|
def test_delete_is_async(self):
|
|
"""delete must be an async coroutine on OneDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
assert hasattr(OneDriveBackend, "delete"), (
|
|
"OneDriveBackend must implement delete (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(OneDriveBackend.delete)
|
|
|
|
def test_upload_file_is_async(self):
|
|
"""upload_file must be an async coroutine on OneDriveBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
assert hasattr(OneDriveBackend, "upload_file"), (
|
|
"OneDriveBackend must implement upload_file (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(OneDriveBackend.upload_file)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_hands_off_new_credentials(self):
|
|
"""Successful token refresh returns new credentials for service-layer persistence.
|
|
|
|
CONN-02: _refresh_token must return the new credentials dict (or signal the
|
|
service layer) so encrypted persistence can happen. The adapter itself must
|
|
not encrypt or store — that is the service layer's responsibility.
|
|
|
|
FAILS: _refresh_token return contract not yet established.
|
|
"""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
|
|
backend = self._make_backend(expires_at="2000-01-01T00:00:00") # expired
|
|
|
|
new_access = "new_access_token_13"
|
|
new_refresh = "new_refresh_token_13"
|
|
|
|
import msal
|
|
fake_app = MagicMock()
|
|
fake_app.acquire_token_by_refresh_token.return_value = {
|
|
"access_token": new_access,
|
|
"refresh_token": new_refresh,
|
|
"expires_in": 3600,
|
|
}
|
|
|
|
with patch("msal.PublicClientApplication", return_value=fake_app):
|
|
refreshed = await backend._refresh_token()
|
|
|
|
# _refresh_token must return the new credentials dict for service-layer handoff
|
|
assert refreshed is not None, (
|
|
"_refresh_token must return new credentials dict for CONN-02 persistence handoff"
|
|
)
|
|
assert isinstance(refreshed, dict), (
|
|
"_refresh_token must return a dict of new credentials"
|
|
)
|
|
assert refreshed.get("access_token") == new_access, (
|
|
f"Refreshed access_token mismatch: expected {new_access!r}, got {refreshed.get('access_token')!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_permanent_normalizes_result(self):
|
|
"""OneDrive delete returns normalized {'kind': 'deleted', 'reason': 'permanent'}.
|
|
|
|
D-11: OneDrive does not have a REST trash API — deletes are permanent.
|
|
The result must make this explicit to the service layer / frontend.
|
|
|
|
FAILS: delete method does not exist yet.
|
|
"""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
|
|
backend = self._make_backend()
|
|
|
|
mock_resp = MagicMock()
|
|
mock_resp.status_code = 204
|
|
mock_resp.is_success = True
|
|
|
|
with patch("httpx.AsyncClient") as mock_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client.delete = AsyncMock(return_value=mock_resp)
|
|
mock_cls.return_value = mock_client
|
|
|
|
result = await backend.delete("od_item_id_123")
|
|
|
|
assert isinstance(result, dict), "delete must return a normalized dict"
|
|
assert result.get("kind") == "deleted"
|
|
assert result.get("reason") == "permanent", (
|
|
"OneDrive deletes are permanent — reason must be 'permanent' (D-11)"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nextlink_restricted_to_graph_host(self):
|
|
"""OneDrive nextLink must only be followed to graph.microsoft.com host (SSRF guard).
|
|
|
|
An SSRF attack could inject a nextLink pointing to an internal IP or
|
|
attacker-controlled server. The backend must restrict nextLink to the
|
|
known Graph API host.
|
|
|
|
FAILS: Phase 13 nextLink host guard not yet implemented.
|
|
"""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
import uuid
|
|
|
|
backend = self._make_backend()
|
|
|
|
# Inject a malicious nextLink pointing to an attacker host
|
|
malicious_page1 = {
|
|
"value": [{"id": "od1", "name": "file.txt", "file": {}}],
|
|
"@odata.nextLink": "http://169.254.169.254/latest/meta-data/",
|
|
}
|
|
mock_resp = MagicMock()
|
|
mock_resp.is_success = True
|
|
mock_resp.json.return_value = malicious_page1
|
|
|
|
with patch("httpx.AsyncClient") as mock_cls:
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client.get = AsyncMock(return_value=mock_resp)
|
|
mock_cls.return_value = mock_client
|
|
|
|
result = await backend.list_folder(uuid.uuid4(), uuid.uuid4())
|
|
|
|
# The backend must NOT follow the malicious nextLink
|
|
# Result should be complete=False (truncated at the SSRF boundary)
|
|
# OR the listing should be returned without following the malicious page
|
|
assert mock_client.get.call_count >= 1
|
|
# Verify the second call (if made) went only to graph.microsoft.com
|
|
if mock_client.get.call_count > 1:
|
|
for call_args in mock_client.get.call_args_list[1:]:
|
|
url = call_args[0][0] if call_args[0] else call_args[1].get("url", "")
|
|
assert "graph.microsoft.com" in url or "microsoftonline" in url, (
|
|
f"nextLink follow must be restricted to Graph host — SSRF guard (T-13-04); "
|
|
f"got URL: {url!r}"
|
|
)
|
|
|
|
|
|
class TestNextcloudMutableContract:
|
|
"""Nextcloud — Phase 13 mutable-operation contract (RED).
|
|
|
|
T-13-04: Nextcloud/WebDAV outbound calls must be SSRF-guarded.
|
|
All method assertions will fail until Phase 13 implements the mutable interface.
|
|
"""
|
|
|
|
def _make_backend(self):
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
with patch("webdav3.client.Client"):
|
|
backend = NextcloudBackend.__new__(NextcloudBackend)
|
|
backend._server_url = "https://nc.example.com/remote.php/dav/files/user/"
|
|
backend._username = "testuser"
|
|
backend._client = MagicMock()
|
|
return backend
|
|
|
|
def test_create_folder_is_async(self):
|
|
"""create_folder must be an async coroutine on NextcloudBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
assert hasattr(NextcloudBackend, "create_folder"), (
|
|
"NextcloudBackend must implement create_folder (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(NextcloudBackend.create_folder)
|
|
|
|
def test_rename_is_async(self):
|
|
"""rename must be an async coroutine on NextcloudBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
assert hasattr(NextcloudBackend, "rename"), (
|
|
"NextcloudBackend must implement rename (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(NextcloudBackend.rename)
|
|
|
|
def test_move_is_async(self):
|
|
"""move must be an async coroutine on NextcloudBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
assert hasattr(NextcloudBackend, "move"), (
|
|
"NextcloudBackend must implement move (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(NextcloudBackend.move)
|
|
|
|
def test_delete_is_async(self):
|
|
"""delete must be an async coroutine on NextcloudBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
assert hasattr(NextcloudBackend, "delete"), (
|
|
"NextcloudBackend must implement delete (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(NextcloudBackend.delete)
|
|
|
|
def test_upload_file_is_async(self):
|
|
"""upload_file must be an async coroutine on NextcloudBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
assert hasattr(NextcloudBackend, "upload_file"), (
|
|
"NextcloudBackend must implement upload_file (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(NextcloudBackend.upload_file)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_folder_ssrf_guarded(self):
|
|
"""create_folder must validate target path against SSRF redirect attempts (T-13-04).
|
|
|
|
Nextcloud WebDAV create-folder (MKCOL) responses can include a Location header
|
|
pointing to an internal address. The backend must validate that any redirect
|
|
destination is within the configured server_url host.
|
|
|
|
FAILS: create_folder SSRF guard does not exist yet.
|
|
"""
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
from storage.cloud_utils import validate_cloud_url
|
|
|
|
backend = self._make_backend()
|
|
# The backend must reject any MKCOL path that tries to escape the server_url
|
|
backend._client.mkdir.side_effect = Exception("Internal redirect to http://169.254.169.254/")
|
|
|
|
result = await backend.create_folder(
|
|
parent_ref=None,
|
|
name="../../etc",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
assert isinstance(result, dict), "create_folder must return normalized dict even on error"
|
|
assert result.get("kind") in ("folder", "conflict", "error"), (
|
|
f"Unexpected kind: {result.get('kind')!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_normalizes_result(self):
|
|
"""NextcloudBackend delete returns normalized {'kind': 'deleted', 'reason': 'permanent'}.
|
|
|
|
Nextcloud WebDAV does not expose a standard Trash API — deletes are permanent
|
|
unless the Trash plugin is active and specifically targeted. Default = permanent.
|
|
|
|
FAILS: delete method does not exist yet.
|
|
"""
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
|
|
backend = self._make_backend()
|
|
backend._client.clean.return_value = True
|
|
|
|
result = await backend.delete("remote.php/dav/files/user/report.pdf")
|
|
assert isinstance(result, dict), "delete must return normalized dict"
|
|
assert result.get("kind") == "deleted"
|
|
assert result.get("reason") in ("permanent", "trashed"), (
|
|
f"Expected 'permanent' or 'trashed', got {result.get('reason')!r}"
|
|
)
|
|
|
|
|
|
class TestWebDAVMutableContract:
|
|
"""WebDAV — Phase 13 mutable-operation contract (RED).
|
|
|
|
T-13-04: WebDAV outbound SSRF guard must be preserved during mutations.
|
|
All method assertions will fail until Phase 13 implements the mutable interface.
|
|
"""
|
|
|
|
def _make_backend(self):
|
|
from storage.webdav_backend import WebDAVBackend
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
with patch("webdav3.client.Client"):
|
|
backend = WebDAVBackend.__new__(WebDAVBackend)
|
|
backend._server_url = "https://dav.example.com/dav/"
|
|
backend._client = MagicMock()
|
|
return backend
|
|
|
|
def test_create_folder_is_async(self):
|
|
"""create_folder must be an async coroutine on WebDAVBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.webdav_backend import WebDAVBackend
|
|
assert hasattr(WebDAVBackend, "create_folder"), (
|
|
"WebDAVBackend must implement create_folder (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(WebDAVBackend.create_folder)
|
|
|
|
def test_rename_is_async(self):
|
|
"""rename must be an async coroutine on WebDAVBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.webdav_backend import WebDAVBackend
|
|
assert hasattr(WebDAVBackend, "rename"), (
|
|
"WebDAVBackend must implement rename (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(WebDAVBackend.rename)
|
|
|
|
def test_move_is_async(self):
|
|
"""move must be an async coroutine on WebDAVBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.webdav_backend import WebDAVBackend
|
|
assert hasattr(WebDAVBackend, "move"), (
|
|
"WebDAVBackend must implement move (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(WebDAVBackend.move)
|
|
|
|
def test_delete_is_async(self):
|
|
"""delete must be an async coroutine on WebDAVBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.webdav_backend import WebDAVBackend
|
|
assert hasattr(WebDAVBackend, "delete"), (
|
|
"WebDAVBackend must implement delete (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(WebDAVBackend.delete)
|
|
|
|
def test_upload_file_is_async(self):
|
|
"""upload_file must be an async coroutine on WebDAVBackend (Phase 13).
|
|
|
|
FAILS: method does not exist yet.
|
|
"""
|
|
import inspect
|
|
from storage.webdav_backend import WebDAVBackend
|
|
assert hasattr(WebDAVBackend, "upload_file"), (
|
|
"WebDAVBackend must implement upload_file (Phase 13)"
|
|
)
|
|
assert inspect.iscoroutinefunction(WebDAVBackend.upload_file)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_permanent_normalizes_result(self):
|
|
"""WebDAVBackend delete returns normalized {'kind': 'deleted', 'reason': 'permanent'}.
|
|
|
|
D-11: Generic WebDAV has no trash API — deletes are permanent.
|
|
The confirmation dialog must say so explicitly.
|
|
|
|
FAILS: delete method does not exist yet.
|
|
"""
|
|
from storage.webdav_backend import WebDAVBackend
|
|
|
|
backend = self._make_backend()
|
|
backend._client.clean.return_value = True
|
|
|
|
result = await backend.delete("/dav/report.pdf")
|
|
assert isinstance(result, dict), "delete must return normalized dict"
|
|
assert result.get("kind") == "deleted"
|
|
assert result.get("reason") == "permanent", (
|
|
"WebDAV deletes are permanent — reason must be 'permanent' (D-11)"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_move_validates_destination_stays_on_same_host(self):
|
|
"""WebDAV MOVE destination must be validated to stay on the configured server host.
|
|
|
|
T-13-04: A WebDAV MOVE request accepts a Destination header. The backend must
|
|
ensure the destination URL cannot escape to a different host (SSRF via Destination).
|
|
|
|
FAILS: move SSRF guard does not exist yet.
|
|
"""
|
|
from storage.webdav_backend import WebDAVBackend
|
|
|
|
backend = self._make_backend()
|
|
|
|
# Attempt to move to a destination on a different host
|
|
try:
|
|
result = await backend.move(
|
|
provider_item_id="report.pdf",
|
|
destination_parent_ref="http://169.254.169.254/evil/",
|
|
etag="v1",
|
|
)
|
|
# If it returns (not raises), it must produce a typed error result
|
|
assert isinstance(result, dict)
|
|
assert result.get("kind") == "invalid_destination", (
|
|
f"Expected kind='invalid_destination' for SSRF destination, got {result.get('kind')!r}"
|
|
)
|
|
except (ValueError, Exception) as exc:
|
|
# Raising an exception with the SSRF block is also acceptable
|
|
assert "SSRF" in str(exc) or "host" in str(exc).lower() or "invalid" in str(exc).lower(), (
|
|
f"SSRF move rejection must mention 'SSRF', 'host', or 'invalid', got: {exc}"
|
|
)
|
|
|
|
def test_no_direct_cloud_items_writes(self):
|
|
"""WebDAVBackend mutation methods must not import or write cloud_items directly.
|
|
|
|
Provider-neutral contract: adapters must never write to cloud_items. Only the
|
|
service layer (services.cloud_items) is allowed to reconcile metadata.
|
|
|
|
FAILS: Phase 13 implementation not yet done — this test will pass once
|
|
we verify the import graph doesn't include cloud_items in the backend module.
|
|
"""
|
|
import importlib
|
|
import sys
|
|
# Load the module source to check imports
|
|
spec = importlib.util.find_spec("storage.webdav_backend")
|
|
if spec and spec.origin:
|
|
with open(spec.origin) as f:
|
|
source = f.read()
|
|
assert "from services.cloud_items" not in source and "import cloud_items" not in source, (
|
|
"WebDAVBackend must not import from services.cloud_items — "
|
|
"only the service layer writes cloud metadata (provider-neutral contract)"
|
|
)
|
|
|
|
|
|
# ── Phase 13 Plan 05: Upload conflict semantics and keep-both naming ──────────
|
|
|
|
|
|
class TestUploadConflictSemantics:
|
|
"""TDD RED — Plan 05 Task 1: typed conflict and retryable-error upload results.
|
|
|
|
Verifies that all four provider adapters normalize upload outcomes into the
|
|
shared typed vocabulary:
|
|
- success → {kind: 'uploaded', reason: 'created', provider_item_id, name, parent_ref, size}
|
|
- conflict → {kind: 'conflict', reason: 'name_collision', existing_name}
|
|
- transient → {kind: 'offline', reason: 'provider_offline'}
|
|
- reauth → {kind: 'reauth_required', reason: 'token_expired'|'invalid_grant'}
|
|
|
|
Keep-both naming (counter before extension) is tested on the shared utility
|
|
and the upload route; the adapter layer only returns 'conflict' — the
|
|
naming decision is the service layer's responsibility.
|
|
|
|
D-03, D-04, T-13-16, T-13-18 (SSRF guard preserved in WebDAV mutable path).
|
|
"""
|
|
|
|
# ── Google Drive upload ────────────────────────────────────────────────────
|
|
|
|
def _make_google_backend(self):
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
return GoogleDriveBackend({
|
|
"access_token": "gd_tok_13",
|
|
"refresh_token": "gd_ref_13",
|
|
"token_uri": "https://oauth2.googleapis.com/token",
|
|
"client_id": "client_id_13",
|
|
"client_secret": "client_secret_13",
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_google_drive_upload_success_returns_typed_result(self):
|
|
"""GoogleDriveBackend.upload_file success returns typed uploaded result.
|
|
|
|
D-03: No silent overwrite — callers must pre-check; adapter must return
|
|
typed 'uploaded' result dict so service layer can proceed with reconcile.
|
|
"""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
from storage.cloud_base import MUT_KIND_UPLOADED
|
|
|
|
backend = self._make_google_backend()
|
|
fake_service = MagicMock()
|
|
fake_service.files().create().execute.return_value = {
|
|
"id": "gd_new_file_id",
|
|
"name": "report.pdf",
|
|
"size": "1024",
|
|
}
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
|
|
result = await backend.upload_file(
|
|
parent_ref=None,
|
|
filename="report.pdf",
|
|
content=b"%PDF-1.4 fake",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict), "upload_file must return a dict"
|
|
assert result.get("kind") == MUT_KIND_UPLOADED, (
|
|
f"Expected kind='uploaded', got {result.get('kind')!r}"
|
|
)
|
|
assert result.get("provider_item_id") == "gd_new_file_id"
|
|
assert result.get("name") == "report.pdf"
|
|
assert "size" in result, "Result must include size"
|
|
# Security: credentials must never appear in the result dict
|
|
for forbidden in ("access_token", "refresh_token", "credentials_enc"):
|
|
assert forbidden not in result, (
|
|
f"upload result must not expose {forbidden!r} (T-13-17)"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_google_drive_upload_transient_error_returns_offline_kind(self):
|
|
"""GoogleDriveBackend.upload_file on provider 503 returns kind='offline'.
|
|
|
|
D-04: A transient provider error must pause the queue for user retry,
|
|
not silently skip. The result must be typed so the service can present
|
|
the 'Retry / Skip / Cancel all' dialog.
|
|
"""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
from storage.cloud_base import MUT_KIND_OFFLINE
|
|
from googleapiclient.errors import HttpError
|
|
|
|
backend = self._make_google_backend()
|
|
fake_service = MagicMock()
|
|
fake_resp = MagicMock()
|
|
fake_resp.status = 503
|
|
fake_service.files().create().execute.side_effect = HttpError(
|
|
resp=fake_resp, content=b'{"error":{"message":"Service Unavailable"}}'
|
|
)
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
|
|
result = await backend.upload_file(
|
|
parent_ref=None,
|
|
filename="report.pdf",
|
|
content=b"data",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict), "upload_file must return normalized dict on error"
|
|
assert result.get("kind") in (MUT_KIND_OFFLINE, "error"), (
|
|
f"Expected kind='offline' or 'error' for transient failure, got {result.get('kind')!r}"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_google_drive_upload_auth_error_returns_reauth_kind(self):
|
|
"""GoogleDriveBackend.upload_file on 401 returns kind='reauth_required'.
|
|
|
|
CONN-02: Token expiry during upload must surface as reauth so the service
|
|
layer can hand off refreshed credentials and retry.
|
|
"""
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
from storage.cloud_base import MUT_KIND_REAUTH
|
|
from googleapiclient.errors import HttpError
|
|
|
|
backend = self._make_google_backend()
|
|
fake_service = MagicMock()
|
|
fake_resp = MagicMock()
|
|
fake_resp.status = 401
|
|
fake_service.files().create().execute.side_effect = HttpError(
|
|
resp=fake_resp, content=b'{"error":{"message":"Invalid Credentials"}}'
|
|
)
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
|
|
result = await backend.upload_file(
|
|
parent_ref=None,
|
|
filename="report.pdf",
|
|
content=b"data",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict), "upload_file must return normalized dict on auth error"
|
|
assert result.get("kind") == MUT_KIND_REAUTH, (
|
|
f"Expected kind='reauth_required' for 401, got {result.get('kind')!r}"
|
|
)
|
|
|
|
# ── OneDrive upload ────────────────────────────────────────────────────────
|
|
|
|
def _make_onedrive_backend(self):
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
return OneDriveBackend({
|
|
"access_token": "od_tok_13",
|
|
"refresh_token": "od_ref_13",
|
|
"expires_at": "2099-01-01T00:00:00",
|
|
})
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_onedrive_upload_success_returns_typed_result(self):
|
|
"""OneDriveBackend.upload_file success returns typed uploaded result.
|
|
|
|
D-03: Uses conflictBehavior=fail so server-detected conflicts return
|
|
the conflict kind rather than silently overwriting.
|
|
"""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
from storage.cloud_base import MUT_KIND_UPLOADED
|
|
|
|
backend = self._make_onedrive_backend()
|
|
session_resp = MagicMock()
|
|
session_resp.is_success = True
|
|
session_resp.json.return_value = {"uploadUrl": "https://upload.microsoft.com/session123"}
|
|
|
|
chunk_resp = MagicMock()
|
|
chunk_resp.status_code = 201
|
|
chunk_resp.is_success = True
|
|
chunk_resp.json.return_value = {"id": "od_new_file_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.post = AsyncMock(return_value=session_resp)
|
|
mock_client.put = AsyncMock(return_value=chunk_resp)
|
|
mock_cls.return_value = mock_client
|
|
|
|
result = await backend.upload_file(
|
|
parent_ref=None,
|
|
filename="doc.pdf",
|
|
content=b"%PDF-1.4",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict), "upload_file must return a dict"
|
|
assert result.get("kind") == MUT_KIND_UPLOADED, (
|
|
f"Expected kind='uploaded', got {result.get('kind')!r}"
|
|
)
|
|
assert result.get("provider_item_id") == "od_new_file_id"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_onedrive_upload_conflict_returns_typed_conflict_kind(self):
|
|
"""OneDriveBackend.upload_file 409 on session create returns kind='conflict'.
|
|
|
|
D-03: conflictBehavior=fail means OneDrive returns 409 on same-name file.
|
|
The adapter must translate this into the typed conflict result — not raise.
|
|
"""
|
|
from storage.onedrive_backend import OneDriveBackend
|
|
from storage.cloud_base import MUT_KIND_CONFLICT, MUT_REASON_NAME_COLLISION
|
|
|
|
backend = self._make_onedrive_backend()
|
|
session_resp = MagicMock()
|
|
session_resp.is_success = False
|
|
session_resp.status_code = 409
|
|
|
|
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.post = AsyncMock(return_value=session_resp)
|
|
mock_cls.return_value = mock_client
|
|
|
|
result = await backend.upload_file(
|
|
parent_ref=None,
|
|
filename="existing.pdf",
|
|
content=b"data",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict)
|
|
assert result.get("kind") == MUT_KIND_CONFLICT, (
|
|
f"Expected kind='conflict' for 409, got {result.get('kind')!r}"
|
|
)
|
|
assert result.get("reason") == MUT_REASON_NAME_COLLISION
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_onedrive_upload_never_writes_cloud_items(self):
|
|
"""OneDriveBackend.upload_file must not import or call cloud_items service.
|
|
|
|
T-13-12: Only the service layer writes to cloud_items. Provider adapters
|
|
must never call reconcile_cloud_listing or upsert_cloud_item directly.
|
|
"""
|
|
import importlib
|
|
spec = importlib.util.find_spec("storage.onedrive_backend")
|
|
if spec and spec.origin:
|
|
with open(spec.origin) as f:
|
|
source = f.read()
|
|
assert "from services.cloud_items" not in source, (
|
|
"OneDriveBackend must not import from services.cloud_items (T-13-12)"
|
|
)
|
|
assert "import cloud_items" not in source, (
|
|
"OneDriveBackend must not import cloud_items directly (T-13-12)"
|
|
)
|
|
|
|
# ── WebDAV upload ──────────────────────────────────────────────────────────
|
|
|
|
def _make_webdav_backend(self):
|
|
from storage.webdav_backend import WebDAVBackend
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
with patch("webdav3.client.Client"):
|
|
backend = WebDAVBackend.__new__(WebDAVBackend)
|
|
backend._server_url = "https://dav.example.com/dav/"
|
|
backend._client = MagicMock()
|
|
return backend
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webdav_upload_success_returns_typed_result(self):
|
|
"""WebDAVBackend.upload_file success returns typed uploaded result."""
|
|
from storage.webdav_backend import WebDAVBackend
|
|
from storage.cloud_base import MUT_KIND_UPLOADED
|
|
|
|
backend = self._make_webdav_backend()
|
|
backend._client.upload_to.return_value = None
|
|
|
|
result = await backend.upload_file(
|
|
parent_ref="Documents",
|
|
filename="report.pdf",
|
|
content=b"%PDF-1.4",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict)
|
|
assert result.get("kind") == MUT_KIND_UPLOADED, (
|
|
f"Expected kind='uploaded', got {result.get('kind')!r}"
|
|
)
|
|
assert result.get("provider_item_id") == "Documents/report.pdf"
|
|
assert result.get("name") == "report.pdf"
|
|
assert result.get("size") == len(b"%PDF-1.4")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webdav_upload_provider_error_returns_error_kind(self):
|
|
"""WebDAVBackend.upload_file on write failure returns typed error result."""
|
|
from storage.webdav_backend import WebDAVBackend
|
|
from storage.cloud_base import MUT_KIND_ERROR, MUT_KIND_OFFLINE
|
|
|
|
backend = self._make_webdav_backend()
|
|
backend._client.upload_to.side_effect = Exception("connection timeout")
|
|
|
|
result = await backend.upload_file(
|
|
parent_ref="Documents",
|
|
filename="report.pdf",
|
|
content=b"data",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(),
|
|
user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict)
|
|
# Connection timeout maps to offline
|
|
assert result.get("kind") in (MUT_KIND_OFFLINE, MUT_KIND_ERROR), (
|
|
f"Unexpected kind: {result.get('kind')!r}"
|
|
)
|
|
|
|
# ── Cross-provider contract ────────────────────────────────────────────────
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider", ["google_drive", "onedrive", "webdav", "nextcloud"])
|
|
async def test_upload_result_never_contains_credentials(self, provider):
|
|
"""Upload results from all providers must never expose credentials (T-13-17)."""
|
|
forbidden_keys = {
|
|
"access_token", "refresh_token", "credentials_enc",
|
|
"client_secret", "client_id", "password",
|
|
}
|
|
|
|
if provider == "google_drive":
|
|
from storage.google_drive_backend import GoogleDriveBackend
|
|
from googleapiclient.errors import HttpError
|
|
backend = self._make_google_backend()
|
|
fake_service = MagicMock()
|
|
fake_resp = MagicMock()
|
|
fake_resp.status = 401
|
|
fake_service.files().create().execute.side_effect = HttpError(
|
|
resp=fake_resp, content=b"Unauthorized"
|
|
)
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
result = await backend.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
elif provider == "onedrive":
|
|
backend = self._make_onedrive_backend()
|
|
err_resp = MagicMock()
|
|
err_resp.is_success = False
|
|
err_resp.status_code = 401
|
|
with patch("httpx.AsyncClient") as mc:
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client.post = AsyncMock(return_value=err_resp)
|
|
mc.return_value = mock_client
|
|
result = await backend.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
elif provider == "webdav":
|
|
backend = self._make_webdav_backend()
|
|
backend._client.upload_to.side_effect = Exception("server error")
|
|
result = await backend.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
elif provider == "nextcloud":
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
with patch("webdav3.client.Client"):
|
|
nc = NextcloudBackend.__new__(NextcloudBackend)
|
|
nc._server_url = "https://nc.example.com/"
|
|
nc._client = MagicMock()
|
|
nc._client.upload_to.side_effect = Exception("error")
|
|
result = await nc.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict), f"{provider}: upload_file must return dict"
|
|
leaked = forbidden_keys.intersection(result.keys())
|
|
assert not leaked, (
|
|
f"{provider}: upload result exposes credential keys: {leaked} (T-13-17)"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("provider", ["google_drive", "onedrive", "webdav", "nextcloud"])
|
|
async def test_upload_result_kind_in_known_kinds(self, provider):
|
|
"""Upload results from all providers must use only known MUT_KIND_* constants."""
|
|
from storage.cloud_base import MUT_KINDS
|
|
|
|
if provider == "google_drive":
|
|
backend = self._make_google_backend()
|
|
fake_service = MagicMock()
|
|
fake_service.files().create().execute.return_value = {
|
|
"id": "gd_id", "name": "f.pdf", "size": "8"
|
|
}
|
|
backend._get_service = MagicMock(return_value=fake_service)
|
|
result = await backend.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
elif provider == "onedrive":
|
|
backend = self._make_onedrive_backend()
|
|
sess_r = MagicMock()
|
|
sess_r.is_success = True
|
|
sess_r.json.return_value = {"uploadUrl": "https://upload.microsoft.com/sess"}
|
|
chunk_r = MagicMock()
|
|
chunk_r.status_code = 201
|
|
chunk_r.is_success = True
|
|
chunk_r.json.return_value = {"id": "od_id"}
|
|
with patch("httpx.AsyncClient") as mc:
|
|
mock_client = AsyncMock()
|
|
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
|
|
mock_client.__aexit__ = AsyncMock(return_value=None)
|
|
mock_client.post = AsyncMock(return_value=sess_r)
|
|
mock_client.put = AsyncMock(return_value=chunk_r)
|
|
mc.return_value = mock_client
|
|
result = await backend.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
elif provider == "webdav":
|
|
backend = self._make_webdav_backend()
|
|
backend._client.upload_to.return_value = None
|
|
result = await backend.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
elif provider == "nextcloud":
|
|
from storage.nextcloud_backend import NextcloudBackend
|
|
with patch("storage.webdav_backend.validate_cloud_url"):
|
|
with patch("webdav3.client.Client"):
|
|
nc = NextcloudBackend.__new__(NextcloudBackend)
|
|
nc._server_url = "https://nc.example.com/"
|
|
nc._client = MagicMock()
|
|
nc._client.upload_to.return_value = None
|
|
result = await nc.upload_file(
|
|
parent_ref=None, filename="f.pdf", content=b"x",
|
|
content_type="application/pdf",
|
|
connection_id=_uuid.uuid4(), user_id=_uuid.uuid4(),
|
|
)
|
|
|
|
assert isinstance(result, dict), f"{provider}: upload_file must return dict"
|
|
assert result.get("kind") in MUT_KINDS, (
|
|
f"{provider}: upload kind {result.get('kind')!r} is not a known MUT_KIND_* constant"
|
|
)
|
|
|
|
|
|
class TestKeepBothNaming:
|
|
"""TDD RED — Plan 05 Task 1: keep-both naming format (counter before extension).
|
|
|
|
D-05: Keep-both collision names place the counter BEFORE the file extension:
|
|
'Report.pdf' → 'Report (1).pdf'
|
|
'Report (1).pdf' → 'Report (2).pdf'
|
|
'archive.tar.gz' → 'archive (1).tar.gz'
|
|
'README' → 'README (1)'
|
|
|
|
The naming helper lives in services.cloud_operations so the upload queue can
|
|
use it consistently across all providers.
|
|
"""
|
|
|
|
def test_keep_both_naming_module_exists(self):
|
|
"""services.cloud_operations must expose a keep-both naming helper."""
|
|
import services.cloud_operations as svc
|
|
assert hasattr(svc, "keep_both_name"), (
|
|
"services.cloud_operations must expose keep_both_name(filename, counter) helper"
|
|
)
|
|
|
|
def test_keep_both_basic_pdf(self):
|
|
"""Report.pdf + counter 1 → Report (1).pdf (counter before extension)."""
|
|
from services.cloud_operations import keep_both_name
|
|
assert keep_both_name("Report.pdf", 1) == "Report (1).pdf"
|
|
|
|
def test_keep_both_counter_increments(self):
|
|
"""Report.pdf + counter 2 → Report (2).pdf."""
|
|
from services.cloud_operations import keep_both_name
|
|
assert keep_both_name("Report.pdf", 2) == "Report (2).pdf"
|
|
|
|
def test_keep_both_no_extension(self):
|
|
"""Files without extension: README → README (1)."""
|
|
from services.cloud_operations import keep_both_name
|
|
assert keep_both_name("README", 1) == "README (1)"
|
|
|
|
def test_keep_both_compound_extension(self):
|
|
"""Compound extension: archive.tar.gz → archive (1).tar.gz.
|
|
|
|
Counter goes before the first dot, not the last.
|
|
'archive.tar.gz' stem='archive', suffix='.tar.gz'.
|
|
"""
|
|
from services.cloud_operations import keep_both_name
|
|
result = keep_both_name("archive.tar.gz", 1)
|
|
# The counter must go before the extension (not appended after)
|
|
assert result.endswith(".tar.gz"), (
|
|
f"Compound extension must be preserved: got {result!r}"
|
|
)
|
|
assert "(1)" in result, f"Counter must be present: got {result!r}"
|
|
|
|
def test_keep_both_preserves_existing_counter(self):
|
|
"""If name already has a counter suffix, bump the number."""
|
|
from services.cloud_operations import keep_both_name
|
|
# Should produce the passed counter value regardless
|
|
assert keep_both_name("Report (1).pdf", 2) == "Report (1) (2).pdf"
|
|
|
|
def test_keep_both_spaces_in_name(self):
|
|
"""Filenames with spaces are handled correctly."""
|
|
from services.cloud_operations import keep_both_name
|
|
assert keep_both_name("My Document.docx", 1) == "My Document (1).docx"
|