Multiple folder move added
This commit is contained in:
kodi
@@ -26,6 +26,18 @@ class FailingDeleteFilesystemAdapter(FilesystemAdapter):
|
||||
raise OSError("forced delete failure")
|
||||
|
||||
|
||||
class FailingBatchFilesystemAdapter(FilesystemAdapter):
|
||||
def move_file(self, source: str, destination: str) -> None:
|
||||
if Path(source).name == "fail-file.txt":
|
||||
raise OSError("forced batch move failure")
|
||||
super().move_file(source, destination)
|
||||
|
||||
def move_directory(self, source: str, destination: str) -> None:
|
||||
if Path(source).name == "fail-dir":
|
||||
raise OSError("forced batch move failure")
|
||||
super().move_directory(source, destination)
|
||||
|
||||
|
||||
class MoveApiGoldenTest(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.temp_dir = tempfile.TemporaryDirectory()
|
||||
@@ -156,6 +168,193 @@ class MoveApiGoldenTest(unittest.TestCase):
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertEqual(response.json()["error"]["code"], "invalid_request")
|
||||
|
||||
def test_move_batch_same_root_directories_success(self) -> None:
|
||||
first = self.root1 / "first-dir"
|
||||
second = self.root1 / "second-dir"
|
||||
first.mkdir()
|
||||
second.mkdir()
|
||||
(first / "a.txt").write_text("a", encoding="utf-8")
|
||||
(second / "b.txt").write_text("b", encoding="utf-8")
|
||||
target = self.root1 / "target"
|
||||
target.mkdir()
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/first-dir", "storage1/second-dir"],
|
||||
"destination_base": "storage1/target",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 202)
|
||||
detail = self._wait_task(response.json()["task_id"])
|
||||
self.assertEqual(detail["status"], "completed")
|
||||
self.assertEqual(detail["done_items"], 2)
|
||||
self.assertEqual(detail["total_items"], 2)
|
||||
self.assertTrue((target / "first-dir").is_dir())
|
||||
self.assertTrue((target / "second-dir").is_dir())
|
||||
self.assertFalse(first.exists())
|
||||
self.assertFalse(second.exists())
|
||||
|
||||
def test_move_batch_same_root_mixed_files_and_directories_success(self) -> None:
|
||||
source_file = self.root1 / "one.txt"
|
||||
source_file.write_text("x", encoding="utf-8")
|
||||
source_dir = self.root1 / "dir-a"
|
||||
source_dir.mkdir()
|
||||
(source_dir / "nested.txt").write_text("y", encoding="utf-8")
|
||||
target = self.root1 / "target"
|
||||
target.mkdir()
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/one.txt", "storage1/dir-a"],
|
||||
"destination_base": "storage1/target",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 202)
|
||||
detail = self._wait_task(response.json()["task_id"])
|
||||
self.assertEqual(detail["status"], "completed")
|
||||
self.assertEqual(detail["done_items"], 2)
|
||||
self.assertEqual(detail["total_items"], 2)
|
||||
self.assertTrue((target / "one.txt").exists())
|
||||
self.assertTrue((target / "dir-a").is_dir())
|
||||
self.assertFalse(source_file.exists())
|
||||
self.assertFalse(source_dir.exists())
|
||||
|
||||
def test_move_batch_cross_root_directories_blocked(self) -> None:
|
||||
first = self.root1 / "first-dir"
|
||||
second = self.root1 / "second-dir"
|
||||
first.mkdir()
|
||||
second.mkdir()
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/first-dir", "storage1/second-dir"],
|
||||
"destination_base": "storage2",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertEqual(response.json()["error"]["code"], "invalid_request")
|
||||
|
||||
def test_move_batch_mixed_root_selection_blocked(self) -> None:
|
||||
first = self.root1 / "first-dir"
|
||||
second = self.root2 / "other-dir"
|
||||
first.mkdir()
|
||||
second.mkdir()
|
||||
target = self.root1 / "target"
|
||||
target.mkdir()
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/first-dir", "storage2/other-dir"],
|
||||
"destination_base": "storage1/target",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertEqual(response.json()["error"]["code"], "invalid_request")
|
||||
|
||||
def test_move_batch_destination_exists_blocked(self) -> None:
|
||||
first = self.root1 / "first-dir"
|
||||
second = self.root1 / "second-dir"
|
||||
first.mkdir()
|
||||
second.mkdir()
|
||||
target = self.root1 / "target"
|
||||
target.mkdir()
|
||||
(target / "second-dir").mkdir()
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/first-dir", "storage1/second-dir"],
|
||||
"destination_base": "storage1/target",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 409)
|
||||
self.assertEqual(response.json()["error"]["code"], "already_exists")
|
||||
|
||||
def test_move_batch_destination_inside_source_blocked(self) -> None:
|
||||
first = self.root1 / "first-dir"
|
||||
first.mkdir()
|
||||
(first / "child").mkdir()
|
||||
second = self.root1 / "second-dir"
|
||||
second.mkdir()
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/first-dir", "storage1/second-dir"],
|
||||
"destination_base": "storage1/first-dir/child",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 400)
|
||||
self.assertEqual(response.json()["error"]["code"], "invalid_request")
|
||||
|
||||
def test_move_batch_symlink_source_blocked(self) -> None:
|
||||
real_dir = self.root1 / "real-dir"
|
||||
real_dir.mkdir()
|
||||
symlink = self.root1 / "dir-link"
|
||||
symlink.symlink_to(real_dir, target_is_directory=True)
|
||||
other = self.root1 / "other-dir"
|
||||
other.mkdir()
|
||||
target = self.root1 / "target"
|
||||
target.mkdir()
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/dir-link", "storage1/other-dir"],
|
||||
"destination_base": "storage1/target",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 409)
|
||||
self.assertEqual(response.json()["error"]["code"], "type_conflict")
|
||||
|
||||
def test_move_batch_runtime_io_error_failed_task_shape(self) -> None:
|
||||
first = self.root1 / "ok-dir"
|
||||
first.mkdir()
|
||||
second = self.root1 / "fail-dir"
|
||||
second.mkdir()
|
||||
target = self.root1 / "target"
|
||||
target.mkdir()
|
||||
|
||||
path_guard = PathGuard({"storage1": str(self.root1), "storage2": str(self.root2)})
|
||||
self._set_services(path_guard=path_guard, filesystem=FailingBatchFilesystemAdapter())
|
||||
|
||||
response = self._request(
|
||||
"POST",
|
||||
"/api/files/move",
|
||||
{
|
||||
"sources": ["storage1/ok-dir", "storage1/fail-dir"],
|
||||
"destination_base": "storage1/target",
|
||||
},
|
||||
)
|
||||
|
||||
self.assertEqual(response.status_code, 202)
|
||||
detail = self._wait_task(response.json()["task_id"])
|
||||
self.assertEqual(detail["status"], "failed")
|
||||
self.assertEqual(detail["error_code"], "io_error")
|
||||
self.assertEqual(detail["done_items"], 1)
|
||||
self.assertEqual(detail["total_items"], 2)
|
||||
self.assertEqual(detail["failed_item"], str(second))
|
||||
self.assertTrue((target / "ok-dir").is_dir())
|
||||
self.assertTrue(second.exists())
|
||||
|
||||
def test_move_source_not_found(self) -> None:
|
||||
response = self._request(
|
||||
"POST",
|
||||
|
||||
Reference in New Issue
Block a user