Tests: WebDAV resource ops, drive create/update, validate/stats

- InternxtDAVResource: delete (real/pending/no-uuid/error), move_recursive
  (different folder/same folder/move+rename/pending/error), copy_move
  (root/subfolder/error).
- InternxtDAVCollection: move_recursive (different parent/rename/uuid-
  resolve/error), copy_recursive (not implemented), date accessors.
- drive_service.create_folder: default-to-root-uuid, explicit parent,
  timestamps included/omitted, parent cache update when present, no
  cache mutation when absent, raises when no root id.
- drive_service.update_file: full encrypt+upload+replace cycle, parent
  cache invalidation, error wrapping.
- validate_upload_sources: missing/dir-without-recursive/dir-with-
  recursive/oversized/mixed.
- get_upload_statistics: single-file/recursive/non-recursive/empty/
  nonexistent.

Coverage: 76% -> 83%; webdav_provider: 70% -> 84%; drive.py: 78% -> 87%.
Tests: 439 -> 480.

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`            |     85% |
-| `services/drive.py`             |     78% |
+| `services/drive.py`             |     87% |
 | `services/network_utils.py`     |     90% |
-| `services/webdav_provider.py`   |     70% |
+| `services/webdav_provider.py`   |     84% |
 | `services/webdav_server.py`     |     58% |
 | `utils/api.py`                  |     74% |
-| **Total**                       |    **76%** |
+| **Total**                       |    **83%** |
 
-(Total tests: **439** passing in ~3 seconds.)
+(Total tests: **480** passing in ~5 seconds.)

readme.md

@@ -270,7 +270,7 @@ internxt-python/
 │   └── webdav_server.py      # WebDAV server management
 ├── utils/
 │   └── api.py                # HTTP API client
-├── tests/                    # Pytest suite (~440 tests, 76% coverage)
+├── tests/                    # Pytest suite (~480 tests, 83% 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_create_update.py

@@ -0,0 +1,202 @@
+"""Tests for drive_service.create_folder (with cache-update + timestamps)
+and drive_service.update_file (the WebDAV PUT-on-existing path).
+"""
+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
+
+
+@pytest.fixture
+def fake_creds():
+    return {
+        'user': {
+            'rootFolderId': 'root-uuid',
+            'bucket': '00' * 12,
+            'mnemonic': ('abandon abandon abandon abandon abandon abandon '
+                         'abandon abandon abandon abandon abandon about'),
+            'bridgeUser': 'u@example.com',
+            'userId': 'u-42',
+        },
+    }
+
+
+# ---------- create_folder ----------
+
+def test_create_folder_uses_root_when_no_parent_given(fake_creds):
+    """If parent_folder_uuid is None, must default to user's rootFolderId."""
+    captured = {}
+    def fake_api_create(payload):
+        captured['payload'] = payload
+        return {'uuid': 'new-uuid', 'plainName': payload['plainName']}
+
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'create_folder',
+                      side_effect=fake_api_create):
+        result = drive_service.create_folder('NewDir')
+
+    assert captured['payload']['parentFolderUuid'] == 'root-uuid'
+    assert captured['payload']['plainName'] == 'NewDir'
+    assert result['uuid'] == 'new-uuid'
+
+
+def test_create_folder_with_explicit_parent(fake_creds):
+    captured = {}
+    def fake_api_create(payload):
+        captured['payload'] = payload
+        return {'uuid': 'new-uuid'}
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'create_folder',
+                      side_effect=fake_api_create):
+        drive_service.create_folder('Sub', parent_folder_uuid='custom-parent')
+    assert captured['payload']['parentFolderUuid'] == 'custom-parent'
+
+
+def test_create_folder_includes_timestamps_when_provided(fake_creds):
+    captured = {}
+    def fake_api_create(payload):
+        captured['payload'] = payload
+        return {'uuid': 'new-uuid'}
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'create_folder',
+                      side_effect=fake_api_create):
+        drive_service.create_folder(
+            'Dated', parent_folder_uuid='p',
+            creation_time='2025-01-01T00:00:00Z',
+            modification_time='2025-06-01T00:00:00Z',
+        )
+    assert captured['payload']['creationTime'] == '2025-01-01T00:00:00Z'
+    assert captured['payload']['modificationTime'] == '2025-06-01T00:00:00Z'
+
+
+def test_create_folder_omits_timestamps_when_none(fake_creds):
+    captured = {}
+    def fake_api_create(payload):
+        captured['payload'] = payload
+        return {'uuid': 'new-uuid'}
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'create_folder',
+                      side_effect=fake_api_create):
+        drive_service.create_folder('Plain', parent_folder_uuid='p')
+    assert 'creationTime' not in captured['payload']
+    assert 'modificationTime' not in captured['payload']
+
+
+def test_create_folder_updates_parent_cache_when_present(fake_creds):
+    """If parent is already cached, the new folder appears in subsequent listings
+    without a re-fetch."""
+    drive_service.folder_content_cache['parent-uuid'] = (
+        9999999999.0, {'folders': [], 'files': []},
+    )
+
+    new_folder = {'uuid': 'new-uuid', 'plainName': 'NewDir'}
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'create_folder',
+                      return_value=new_folder):
+        drive_service.create_folder('NewDir', parent_folder_uuid='parent-uuid')
+
+    _, content = drive_service.folder_content_cache['parent-uuid']
+    assert any(f.get('uuid') == 'new-uuid' for f in content['folders'])
+
+
+def test_create_folder_skips_cache_update_when_parent_not_cached(fake_creds):
+    """If parent isn't cached, no cache mutation — the next get_folder_content
+    call will fetch fresh."""
+    new_folder = {'uuid': 'new-uuid', 'plainName': 'NewDir'}
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'create_folder',
+                      return_value=new_folder):
+        drive_service.create_folder('NewDir', parent_folder_uuid='unknown-parent')
+    # Cache stays empty (no entry was created)
+    assert 'unknown-parent' not in drive_service.folder_content_cache
+
+
+def test_create_folder_raises_when_no_root_id():
+    """If creds have no rootFolderId AND no explicit parent → ValueError."""
+    bad_creds = {'user': {}}
+    with patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=bad_creds):
+        with pytest.raises(ValueError, match="No root folder"):
+            drive_service.create_folder('X')
+
+
+# ---------- update_file (WebDAV PUT-on-existing) ----------
+
+def test_update_file_full_cycle(tmp_path, fake_creds):
+    """update_file: read local file → encrypt → start upload → upload chunk →
+    finish upload → replace_file metadata. Verify each step is called."""
+    local = tmp_path / "doc.txt"
+    local.write_bytes(b"updated content")
+
+    with patch.object(drive_service.api, 'get_file_metadata',
+                      return_value={'plainName': 'doc'}), \
+         patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'start_upload',
+                      return_value={'uploads': [{
+                          'index': 0, 'size': 100,
+                          'url': 'https://upload', 'uuid': 'net-uuid',
+                      }]}), \
+         patch.object(drive_service.api, 'upload_chunk') as mock_chunk, \
+         patch.object(drive_service.api, 'finish_upload',
+                      return_value={'id': 'new-net-id'}), \
+         patch.object(drive_service.api, 'replace_file',
+                      return_value={'success': True}) as mock_replace, \
+         patch.object(drive_service, '_clear_parent_cache_for_item'):
+        result = drive_service.update_file('file-uuid', str(local))
+
+    assert result['success'] is True
+    mock_chunk.assert_called_once()
+    # replace_file gets the new network file id and the local file size
+    args, _ = mock_replace.call_args
+    assert args[0] == 'file-uuid'
+    assert args[1]['fileId'] == 'new-net-id'
+    assert args[1]['size'] == len(b"updated content")
+
+
+def test_update_file_clears_parent_cache(tmp_path, fake_creds):
+    """After the update, parent cache must be invalidated so listings refresh."""
+    local = tmp_path / "doc.txt"
+    local.write_bytes(b"x")
+
+    with patch.object(drive_service.api, 'get_file_metadata',
+                      return_value={'plainName': 'doc'}), \
+         patch.object(drive_service.auth, 'get_auth_details',
+                      return_value=fake_creds), \
+         patch.object(drive_service.api, 'start_upload',
+                      return_value={'uploads': [{
+                          'index': 0, 'size': 1,
+                          'url': 'u', 'uuid': 'n',
+                      }]}), \
+         patch.object(drive_service.api, 'upload_chunk'), \
+         patch.object(drive_service.api, 'finish_upload',
+                      return_value={'id': 'nid'}), \
+         patch.object(drive_service.api, 'replace_file',
+                      return_value={}), \
+         patch.object(drive_service, '_clear_parent_cache_for_item') as mock_clear:
+        drive_service.update_file('file-uuid', str(local))
+
+    mock_clear.assert_called_once_with('file-uuid', 'file')
+
+
+def test_update_file_wraps_errors():
+    with patch.object(drive_service.api, 'get_file_metadata',
+                      side_effect=ConnectionError("net")):
+        with pytest.raises(Exception, match="Failed to update file"):
+            drive_service.update_file('file-uuid', '/tmp/nope')

tests/test_drive_validate_stats.py

@@ -0,0 +1,134 @@
+"""Tests for drive_service validation/statistics helpers:
+validate_upload_sources, get_upload_statistics.
+"""
+from pathlib import Path
+
+
+from services.drive import drive_service
+
+
+# ---------- validate_upload_sources ----------
+
+def test_validate_returns_valid_paths_for_real_files(tmp_path):
+    a = tmp_path / "a.txt"
+    a.write_bytes(b"hello")
+    b = tmp_path / "b.txt"
+    b.write_bytes(b"world")
+    valid, errors = drive_service.validate_upload_sources([str(a), str(b)])
+    assert len(valid) == 2
+    assert errors == []
+
+
+def test_validate_reports_missing_source(tmp_path):
+    valid, errors = drive_service.validate_upload_sources([str(tmp_path / "ghost.txt")])
+    assert valid == []
+    assert any("not found" in e.lower() for e in errors)
+
+
+def test_validate_rejects_directory_without_recursive(tmp_path):
+    sub = tmp_path / "subdir"
+    sub.mkdir()
+    valid, errors = drive_service.validate_upload_sources([str(sub)], recursive=False)
+    assert valid == []
+    assert any("recursive" in e.lower() for e in errors)
+
+
+def test_validate_accepts_directory_with_recursive(tmp_path):
+    sub = tmp_path / "subdir"
+    sub.mkdir()
+    valid, errors = drive_service.validate_upload_sources([str(sub)], recursive=True)
+    assert len(valid) == 1
+    assert errors == []
+
+
+def test_validate_skips_oversized_files(tmp_path, monkeypatch):
+    """Files over the 20 GB limit are reported as errors, not in valid list."""
+    big = tmp_path / "big.bin"
+    big.write_bytes(b"x")  # actually small
+
+    # Patch stat to claim huge size
+    real_stat = Path.stat
+    def fake_stat(self, *a, **k):
+        result = real_stat(self, *a, **k)
+        if self == big:
+            # Return a stat result with the file's existing fields but huge size.
+            class Fake:
+                def __init__(self, real, size):
+                    for f in dir(real):
+                        if f.startswith('st_'):
+                            setattr(self, f, getattr(real, f))
+                    self.st_size = size
+            return Fake(result, drive_service.TWENTY_GIGABYTES + 1)
+        return result
+    monkeypatch.setattr(Path, 'stat', fake_stat)
+
+    valid, errors = drive_service.validate_upload_sources([str(big)])
+    assert valid == []
+    assert any("too large" in e.lower() for e in errors)
+
+
+def test_validate_mixes_valid_and_invalid(tmp_path):
+    good = tmp_path / "good.txt"
+    good.write_bytes(b"x")
+    valid, errors = drive_service.validate_upload_sources(
+        [str(good), str(tmp_path / "missing.txt")])
+    assert len(valid) == 1
+    assert len(errors) == 1
+
+
+# ---------- get_upload_statistics ----------
+
+def test_stats_for_single_file(tmp_path):
+    f = tmp_path / "doc.txt"
+    f.write_bytes(b"hello world")
+    stats = drive_service.get_upload_statistics(f)
+    assert stats['total_files'] == 1
+    assert stats['total_size'] == 11
+    assert stats['total_dirs'] == 0
+    assert stats['file_list'] == [f]
+
+
+def test_stats_for_directory_recursive(tmp_path):
+    """Recursive walk counts files at every depth + intermediate dirs."""
+    (tmp_path / "a.txt").write_bytes(b"a" * 100)
+    sub = tmp_path / "sub"
+    sub.mkdir()
+    (sub / "b.txt").write_bytes(b"b" * 200)
+    deeper = sub / "deeper"
+    deeper.mkdir()
+    (deeper / "c.txt").write_bytes(b"c" * 300)
+
+    stats = drive_service.get_upload_statistics(tmp_path, recursive=True)
+    assert stats['total_files'] == 3
+    assert stats['total_size'] == 600
+    assert stats['total_dirs'] == 2  # 'sub' and 'deeper'
+
+
+def test_stats_for_directory_non_recursive_only_top_level(tmp_path):
+    """Without recursive, only direct children are counted."""
+    (tmp_path / "top.txt").write_bytes(b"t" * 50)
+    sub = tmp_path / "sub"
+    sub.mkdir()
+    (sub / "deep.txt").write_bytes(b"d" * 100)
+
+    stats = drive_service.get_upload_statistics(tmp_path, recursive=False)
+    assert stats['total_files'] == 1  # 'top.txt' only
+    assert stats['total_dirs'] == 1
+    assert stats['total_size'] == 50
+
+
+def test_stats_for_empty_directory(tmp_path):
+    sub = tmp_path / "empty"
+    sub.mkdir()
+    stats = drive_service.get_upload_statistics(sub, recursive=True)
+    assert stats['total_files'] == 0
+    assert stats['total_size'] == 0
+    assert stats['total_dirs'] == 0
+    assert stats['file_list'] == []
+
+
+def test_stats_for_nonexistent_path_is_empty(tmp_path):
+    """Non-file, non-dir input yields empty stats (no crash)."""
+    stats = drive_service.get_upload_statistics(tmp_path / "missing")
+    assert stats['total_files'] == 0
+    assert stats['total_size'] == 0