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)
This commit is contained in:
curo1305
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user