From 4efe7c13763c34b8996984598cff708f06fe7625 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Thu, 28 May 2026 21:06:14 +0200 Subject: [PATCH] test(05-03): add RED phase tests for GoogleDriveBackend and OneDriveBackend - 32 failing tests covering all 7 StorageBackend methods on both backends - Verifies CloudConnectionError reason attribute (token_expired / invalid_grant) - Verifies CHUNK_SIZE == 10 MB (Pitfall 6 prevention) - Verifies shared CloudConnectionError import across backends - Verifies _ensure_valid_token skips refresh on non-expired tokens - Verifies _ensure_valid_token raises CloudConnectionError on invalid_grant --- backend/tests/test_cloud_backends.py | 285 +++++++++++++++++++++++++++ 1 file changed, 285 insertions(+) create mode 100644 backend/tests/test_cloud_backends.py diff --git a/backend/tests/test_cloud_backends.py b/backend/tests/test_cloud_backends.py new file mode 100644 index 0000000..c8fe3a6 --- /dev/null +++ b/backend/tests/test_cloud_backends.py @@ -0,0 +1,285 @@ +""" +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 .""" + 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"