curo1305
ff33439f0a
- GoogleDriveBackend: list_folder with full pagination, native doc size=None, get_capabilities - OneDriveBackend: list_folder with @odata.nextLink pagination, get_capabilities - WebDAVBackend: list_folder via PROPFIND with SSRF re-validation, get_capabilities - NextcloudBackend: inherits CloudResourceAdapter from WebDAVBackend - build_cloud_resource_adapter() added to factory - 22 new contract/behavior/pagination/no-byte-download tests (51 total pass)
640 lines
26 KiB
Python
640 lines
26 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 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 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"
|