Add regression tests for per-file diskcache reuse when streaming remote files

Streaming a proxied remote file (RemoteFile) reopened the per-file chunk
diskcache on essentially every read block, which is cheap on fast storage but
pathologically slow on SD cards. Cover:

- the diskcache is constructed a bounded number of times regardless of file
  size (not once per chunk)
- concurrent readers of the same file, and single-read correctness
- the handle self-heals when the chunk directory is evicted and recreated
  mid-stream, including dropping a borrowed handle without closing it
- the self-heal mechanism and Windows-safety invariant cross-platform:
  _open_cache reopens on an incarnation change, and delete() releases the
  handle before removing the directory
- the marker-file fallback used when the filesystem reports no inode
- eviction skips a directory that cannot be removed rather than aborting

The end-to-end deletion flavors are POSIX-only, since Windows forbids removing
a directory whose diskcache is held open; tests release the handle before
removing a directory so it does not block removal there.

Claude-Session: https://claude.ai/code/session_01MnAmUbaCBpL4wBKFYUyCwt

Author: jamalex

kolibri/utils/tests/test_chunked_file.py

@@ -5,6 +5,8 @@
 import tempfile
 import unittest
 
+from mock import patch
+
 from kolibri.utils.file_transfer import CHUNK_SUFFIX
 from kolibri.utils.file_transfer import ChunkedFile
 from kolibri.utils.file_transfer import ChunkedFileDirectoryManager
@@ -48,6 +50,9 @@ def setUp(self):
         self.data = _write_test_data_to_chunked_file(self.chunked_file)
 
     def tearDown(self):
+        # Release the reused diskcache handle before removing the directory, or
+        # the open SQLite file blocks removal on Windows and pollutes later tests.
+        self.chunked_file.close()
         shutil.rmtree(self.chunked_file.chunk_dir, ignore_errors=True)
         shutil.rmtree(self.file_path, ignore_errors=True)
 
@@ -167,7 +172,9 @@ def test_get_missing_chunk_ranges_slice(self):
         self.assertEqual(missing_ranges, expected_ranges)
 
     def test_get_missing_chunk_ranges_slice_no_download(self):
-        # Remove some chunks
+        # Remove some chunks. Release our handle first so the directory can be
+        # removed on Windows (an open SQLite file blocks removal there).
+        self.chunked_file.close()
         shutil.rmtree(self.chunked_file.chunk_dir)
 
         start = self.chunk_size
@@ -182,7 +189,9 @@ def test_get_missing_chunk_ranges_slice_no_download(self):
         self.assertEqual(missing_ranges, expected_ranges)
 
     def test_get_missing_chunk_ranges_slice_no_download_not_chunk_size_ranges(self):
-        # Remove some chunks
+        # Remove some chunks. Release our handle first so the directory can be
+        # removed on Windows (an open SQLite file blocks removal there).
+        self.chunked_file.close()
         shutil.rmtree(self.chunked_file.chunk_dir)
 
         start = self.chunk_size // 3
@@ -274,6 +283,9 @@ def test_finalize_file_md5(self):
         os.remove(self.file_path)
 
     def test_file_removed_by_parallel_process_after_opening(self):
+        # Release our handle so the directory can actually be removed on Windows
+        # (an open SQLite file blocks removal there).
+        self.chunked_file.close()
         shutil.rmtree(self.chunked_file.chunk_dir, ignore_errors=True)
         self.chunked_file._file_size = None
         with self.assertRaises(ChunkedFileDoesNotExist):
@@ -391,6 +403,29 @@ def test_evict_files_more_than_file_size_sum(self):
             sorted([]),
         )
 
+    def test_evict_files_skips_undeletable_directory(self):
+        # If a directory cannot be removed (e.g. a diskcache handle held open
+        # while streaming, which blocks removal on Windows), eviction must skip
+        # it - not count it as freed, and not abort the whole pass.
+        manager = ChunkedFileDirectoryManager(self.base_dir)
+        blocked = sorted(manager._get_chunked_file_dirs())[0]
+        real_rmtree = shutil.rmtree
+
+        def guarded_rmtree(path, *args, **kwargs):
+            if path == blocked:
+                raise OSError("directory in use")
+            return real_rmtree(path, *args, **kwargs)
+
+        with patch(
+            "kolibri.utils.file_transfer.shutil.rmtree", side_effect=guarded_rmtree
+        ):
+            evicted = manager.evict_files(TOTAL_CHUNKED_FILE_SIZE * 3)
+
+        # Two of the three directories are evicted; the blocked one is skipped
+        # and its size is not counted toward the freed total.
+        self.assertEqual(evicted, TOTAL_CHUNKED_FILE_SIZE * 2)
+        self.assertEqual(sorted(manager._get_chunked_file_dirs()), [blocked])
+
     def test_evict_files_exact_file_size(self):
         manager = ChunkedFileDirectoryManager(self.base_dir)
         self.assertEqual(