From f47e36d93d4dd6eeecfb8499f4e81fc29b9af1d1 Mon Sep 17 00:00:00 2001 From: curo1305 Date: Mon, 22 Jun 2026 19:17:13 +0200 Subject: [PATCH] test(13-05): add failing RED tests for upload conflict semantics and keep-both naming - 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) --- backend/tests/test_cloud_backends.py | 517 +++++++++++++++++++++++++++ 1 file changed, 517 insertions(+) diff --git a/backend/tests/test_cloud_backends.py b/backend/tests/test_cloud_backends.py index 2596e84..8662cec 100644 --- a/backend/tests/test_cloud_backends.py +++ b/backend/tests/test_cloud_backends.py @@ -1477,3 +1477,520 @@ class TestWebDAVMutableContract: "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"