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
2026-05-28 21:06:14 +02:00
parent 3b84626da9
commit 4efe7c1376
1 changed files with 285 additions and 0 deletions
@@ -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 <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"