Spaces:
Running
Running
| import unittest | |
| import httpx | |
| from unittest.mock import Mock | |
| from huggingface_hub import RepoFile, RepoFolder | |
| from huggingface_hub.errors import RemoteEntryNotFoundError | |
| from storage_listing import session_files, session_folder_exists | |
| class StorageListingTest(unittest.TestCase): | |
| def test_lists_only_requested_folder_and_excludes_directories(self): | |
| api = Mock() | |
| api.list_repo_tree.return_value = [ | |
| RepoFile(path="sessions/one/videos/camera_1.mp4", size=1, oid="a"), | |
| RepoFolder(path="sessions/one/videos/subdir", oid="b"), | |
| ] | |
| self.assertEqual(session_files(api, "repo", "sessions/one/videos"), | |
| ["sessions/one/videos/camera_1.mp4"]) | |
| api.list_repo_tree.assert_called_once_with( | |
| repo_id="repo", repo_type="dataset", path_in_repo="sessions/one/videos", | |
| recursive=False, expand=False) | |
| api.list_repo_files.assert_not_called() | |
| def test_missing_folder_is_empty_but_other_errors_propagate(self): | |
| api = Mock() | |
| api.list_repo_tree.side_effect = RemoteEntryNotFoundError( | |
| "missing", response=httpx.Response(404, request=httpx.Request("GET", "https://example.com/tree"))) | |
| self.assertEqual(session_files(api, "repo", "folder"), []) | |
| api.list_repo_tree.side_effect = RuntimeError("unavailable") | |
| with self.assertRaises(RuntimeError): | |
| session_files(api, "repo", "folder") | |
| def test_deletion_checks_only_the_session_path(self): | |
| api = Mock() | |
| api.get_paths_info.return_value = [] | |
| self.assertFalse(session_folder_exists(api, "repo", "sessions/one")) | |
| api.get_paths_info.assert_called_once_with( | |
| repo_id="repo", repo_type="dataset", paths=["sessions/one"]) | |
| api.list_repo_files.assert_not_called() | |
| if __name__ == "__main__": | |
| unittest.main() | |