Tests: cross 90% coverage threshold

- drive_service.download_file_by_path: resolve+dispatch, default
  destination filename, folder-rejection.
- upload_single_item edge cases: unreadable file, // path normalization,
  resolve unexpected error continues to upload.
- download_file: creation_time-only branch (no mtime).
- find_files: max_depth=2 (one level deeper), continues searching when
  one subfolder errors.
- WebDAV end_write: large-file disk path timestamp extraction (mtime+
  ctime via st_birthtime fallback to st_ctime), small-file temp file
  timestamp extraction, preserve_timestamps=False skips extraction.
- WebDAVCollection.get_member: returns None on inner exception.
- WebDAVAPIClient._get_isolated_session_without_token_refresh: builds
  client without refresh, thread-local cache, separate threads get
  separate clients.
- get_content error path: BytesIO body contains human-readable error.

Coverage: 88% -> 90%; webdav_provider: 84% -> 91%; drive: 87% -> 89%.
Tests: 540 -> 557.

Author: CrispStrobe

CHANGELOG.md

@@ -117,11 +117,11 @@ All Bandit medium+ findings (was 14, now 0) are resolved or annotated:
 | `config/config.py`              |     85% |
 | `services/auth.py`              |    100% |
 | `services/crypto.py`            |    100% |
-| `services/drive.py`             |     87% |
+| `services/drive.py`             |     89% |
 | `services/network_utils.py`     |     90% |
-| `services/webdav_provider.py`   |     84% |
+| `services/webdav_provider.py`   |     91% |
 | `services/webdav_server.py`     |     83% |
 | `utils/api.py`                  |     98% |
-| **Total**                       |    **88%** |
+| **Total**                       |    **90%** |
 
-(Total tests: **540** passing in ~3 seconds.)
+(Total tests: **557** passing in ~3 seconds.)

readme.md

@@ -270,7 +270,7 @@ internxt-python/
 │   └── webdav_server.py      # WebDAV server management
 ├── utils/
 │   └── api.py                # HTTP API client
-├── tests/                    # Pytest suite (~540 tests, 88% coverage)
+├── tests/                    # Pytest suite (~560 tests, 90% coverage)
 ├── pyproject.toml            # Pytest, coverage, ruff config
 ├── requirements-dev.txt      # Dev/test dependencies
 └── .github/workflows/ci.yml  # Lint + type-check + test on Py 3.10/3.11/3.12

tests/test_drive_misc.py

@@ -0,0 +1,223 @@
+"""Final round of drive_service tests covering remaining gaps:
+download_file_by_path, upload_single_item edge cases (empty target path,
+unreadable file, invalid conflict mode), download_file with creation_time
+warning, search recursion edge cases.
+"""
+from unittest.mock import patch
+
+import pytest
+
+from services.drive import drive_service
+
+
+@pytest.fixture(autouse=True)
+def _reset():
+    drive_service.folder_content_cache.clear()
+    drive_service._mem_reserved = 0
+    yield
+    drive_service.folder_content_cache.clear()
+    drive_service._mem_reserved = 0
+
+
+# ---------- download_file_by_path ----------
+
+def test_download_file_by_path_resolves_then_calls_download_file():
+    resolved = {'type': 'file', 'uuid': 'fid', 'metadata': {},
+                'path': '/Documents/x.pdf'}
+    with patch.object(drive_service, 'resolve_path', return_value=resolved), \
+         patch.object(drive_service, 'download_file',
+                      return_value='/tmp/x.pdf') as mock_dl:
+        out = drive_service.download_file_by_path('/Documents/x.pdf')
+    mock_dl.assert_called_once()
+    assert out == '/tmp/x.pdf'
+
+
+def test_download_file_by_path_default_destination_uses_filename():
+    """If no destination specified, use ./<filename>."""
+    resolved = {'type': 'file', 'uuid': 'fid', 'metadata': {},
+                'path': '/Documents/report.pdf'}
+    captured = {}
+    def fake_dl(uuid, dest, **kw):
+        captured['dest'] = dest
+        return dest
+    with patch.object(drive_service, 'resolve_path', return_value=resolved), \
+         patch.object(drive_service, 'download_file', side_effect=fake_dl):
+        drive_service.download_file_by_path('/Documents/report.pdf')
+    assert captured['dest'].endswith('report.pdf')
+
+
+def test_download_file_by_path_rejects_folder():
+    resolved = {'type': 'folder', 'uuid': 'fid', 'metadata': {}, 'path': '/D'}
+    with patch.object(drive_service, 'resolve_path', return_value=resolved):
+        with pytest.raises(ValueError, match="folder, not a file"):
+            drive_service.download_file_by_path('/D')
+
+
+# ---------- upload_single_item edge cases ----------
+
+def test_upload_single_item_unreadable_file_returns_error(tmp_path):
+    """If stat() raises (e.g., permission denied), return 'error' not crash."""
+    f = tmp_path / "broken.txt"
+    f.write_bytes(b"x")
+
+    with patch('pathlib.Path.is_file', return_value=True), \
+         patch('pathlib.Path.stat', side_effect=PermissionError("denied")):
+        result = drive_service.upload_single_item_with_conflict_handling(
+            f, '/Docs', 'parent-uuid', on_conflict='skip',
+        )
+    assert result == "error"
+
+
+def test_upload_single_item_target_path_normalization(tmp_path):
+    """The full target remote path must always start with exactly one '/'."""
+    f = tmp_path / "doc.txt"
+    f.write_bytes(b"x")
+
+    captured_paths = []
+    def fake_resolve(path):
+        captured_paths.append(path)
+        raise FileNotFoundError(path)
+
+    with patch.object(drive_service, 'resolve_path', side_effect=fake_resolve), \
+         patch.object(drive_service, 'upload_file_to_folder',
+                      return_value={'uuid': 'x'}):
+        # Pass a parent path with NO leading slash → should normalize
+        drive_service.upload_single_item_with_conflict_handling(
+            f, 'Docs/Sub', 'parent-uuid', on_conflict='skip',
+        )
+
+    # The looked-up path must always have exactly one leading slash
+    assert all(p.startswith('/') for p in captured_paths)
+    assert all(not p.startswith('//') for p in captured_paths)
+
+
+def test_upload_single_item_resolve_unexpected_error_continues(tmp_path):
+    """If resolve_path raises something other than FileNotFoundError,
+    log warning and proceed with upload (not skip)."""
+    f = tmp_path / "doc.txt"
+    f.write_bytes(b"x")
+
+    with patch.object(drive_service, 'resolve_path',
+                      side_effect=ConnectionError("flaky API")), \
+         patch.object(drive_service, 'upload_file_to_folder',
+                      return_value={'uuid': 'x'}) as mock_up:
+        result = drive_service.upload_single_item_with_conflict_handling(
+            f, '/Docs', 'parent-uuid', on_conflict='skip',
+        )
+    # Despite the resolve error, upload still proceeded
+    assert result == "uploaded"
+    mock_up.assert_called_once()
+
+
+# ---------- download_file metadata-only branches ----------
+
+def test_download_file_creation_time_warning_branch(tmp_path):
+    """If preserve_timestamps=True and only creation_time is present (no
+    modification_time), the function still completes — it just warns."""
+    fake_creds = {
+        'user': {
+            'bucket': '00' * 12,
+            'mnemonic': ('abandon abandon abandon abandon abandon abandon '
+                         'abandon abandon abandon abandon abandon about'),
+            'bridgeUser': 'u@example.com',
+            'userId': 'u',
+        },
+    }
+    payload = b"content for ctime test"
+    enc, idx_hex = drive_service.crypto.encrypt_stream_internxt_protocol(
+        payload, fake_creds['user']['mnemonic'],
+        fake_creds['user']['bucket'])
+
+    metadata = {
+        'uuid': 'fid', 'bucket': fake_creds['user']['bucket'],
+        'fileId': 'nid', 'size': len(payload),
+        'plainName': 'doc', 'type': 'txt',
+        'creationTime': '2024-01-15T12:00:00Z',
+        # No modificationTime
+    }
+
+    out_dir = tmp_path / "out"
+    out_dir.mkdir()
+
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'get_file_metadata',
+                      return_value=metadata), \
+         patch.object(drive_service.api, 'get_download_links',
+                      return_value={'shards': [{'url': 'u'}], 'index': idx_hex}), \
+         patch.object(drive_service.api, 'download_chunk', return_value=enc):
+        out_path = drive_service.download_file('fid', str(out_dir),
+                                               preserve_timestamps=True)
+    # File downloaded successfully despite no modification time
+    from pathlib import Path
+    assert Path(out_path).exists()
+    assert Path(out_path).read_bytes() == payload
+
+
+# ---------- find_files at deeper depths ----------
+
+def test_find_files_max_depth_2_goes_one_level_deeper():
+    """max_depth=2 → search start folder and one level of subfolders."""
+    tree = {
+        'root-uuid': {
+            'folders': [{'uuid': 'a', 'plainName': 'A'}],
+            'files': [{'uuid': 'top', 'plainName': 'top', 'type': 'pdf', 'size': 1}],
+        },
+        'a': {
+            'folders': [{'uuid': 'b', 'plainName': 'B'}],
+            'files': [{'uuid': 'mid', 'plainName': 'mid', 'type': 'pdf', 'size': 1}],
+        },
+        'b': {
+            'folders': [],
+            'files': [{'uuid': 'deep', 'plainName': 'deep', 'type': 'pdf', 'size': 1}],
+        },
+    }
+    drive_service.folder_content_cache.clear()
+    fake_creds = {'user': {'rootFolderId': 'root-uuid'}}
+
+    def fake_get_content(uuid):
+        return tree.get(uuid, {'folders': [], 'files': []})
+
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service, 'get_folder_content',
+                      side_effect=fake_get_content):
+        results = drive_service.find_files('*.pdf', '/', max_depth=2)
+
+    # Top + mid found, deep filtered out
+    names = sorted(r['display_name'] for r in results)
+    assert names == ['mid.pdf', 'top.pdf']
+
+
+def test_find_files_handles_listing_error_in_subfolder():
+    """If one subfolder fails to list, others must still be searched."""
+    fake_creds = {'user': {'rootFolderId': 'root-uuid'}}
+
+    call_count = {'n': 0}
+    def fake_list(path):
+        call_count['n'] += 1
+        if path == '/':
+            return {
+                'folders': [{'uuid': 'good', 'plainName': 'Good',
+                             'path': '/Good', 'display_name': 'Good'},
+                            {'uuid': 'bad', 'plainName': 'Bad',
+                             'path': '/Bad', 'display_name': 'Bad'}],
+                'files': [],
+            }
+        if path == '/Bad':
+            raise ConnectionError("listing failed")
+        # /Good
+        return {
+            'folders': [],
+            'files': [{'uuid': 'g', 'plainName': 'g', 'type': 'pdf', 'size': 1,
+                       'display_name': 'g.pdf'}],
+        }
+
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service, 'list_folder_with_paths',
+                      side_effect=fake_list):
+        results = drive_service.find_files('*.pdf', '/')
+
+    # Found the file in /Good even though /Bad errored
+    assert any(r['display_name'] == 'g.pdf' for r in results)