PGU-NetFileSyncWin/tests/test_delete_worker.py

import stat
import unittest
from unittest.mock import patch

from pgu4_sync import (
    DELETE_BATCH_ITEMS,
    DELETE_RETRIES,
    DeleteWorker,
    RemoteEntry,
)


class FakeChannel:
    def __init__(self, status):
        self.status = status

    def recv_exit_status(self):
        return self.status


class FakeStream:
    def __init__(self, status=0, data=b""):
        self.channel = FakeChannel(status)
        self.data = data

    def read(self):
        return self.data


class FakeClient:
    def __init__(self, responder=None):
        self.commands = []
        self.responder = responder or (lambda _command: (0, b""))

    def exec_command(self, command):
        self.commands.append(command)
        status, error = self.responder(command)
        return None, FakeStream(status), FakeStream(status, error)

    def close(self):
        return None


class FakeFactory:
    def __init__(self, client=None, error=None):
        self.client = client
        self.error = error

    def open_client(self):
        if self.error is not None:
            raise self.error
        return self.client


def entry(path, is_dir=False):
    mode = stat.S_IFDIR if is_dir else stat.S_IFREG
    return RemoteEntry(path.rsplit("/", 1)[-1], path, 0, 0, mode)


class DeleteWorkerTests(unittest.TestCase):
    def run_worker(self, worker):
        failures = []
        progress = []
        finished = []
        worker.failed.connect(lambda path, message: failures.append((path, message)))
        worker.progress.connect(lambda done, total, path: progress.append((done, total, path)))
        worker.finished.connect(lambda deleted, failed: finished.append((deleted, failed)))
        worker.run()
        return failures, progress, finished

    def test_quotes_paths_and_uses_permanent_rm(self):
        client = FakeClient()
        worker = DeleteWorker(
            FakeFactory(client),
            [entry("/tmp/space dir", True), entry("/tmp/quote'file")],
        )

        failures, progress, finished = self.run_worker(worker)

        self.assertEqual([], failures)
        self.assertEqual([(2, 0)], finished)
        self.assertEqual((2, 2), progress[-1][:2])
        self.assertEqual(1, len(client.commands))
        self.assertTrue(
            client.commands[0].startswith(
                "rm -rf -- '/tmp/space dir' '/tmp/quote'\"'\"'file'"
            )
        )
        self.assertIn("Path still exists after removal", client.commands[0])

    def test_rejects_root_and_noncanonical_paths(self):
        unsafe_paths = [
            "/",
            "//",
            "////",
            "/tmp/../etc",
            "/tmp//item",
            "relative",
            "",
            "/tmp/item\x00tail",
        ]
        factory = FakeFactory(error=AssertionError("SSH must not be opened"))

        for path in unsafe_paths:
            with self.subTest(path=path):
                failures, progress, finished = self.run_worker(
                    DeleteWorker(factory, [entry(path, True)])
                )
                self.assertEqual(1, len(failures))
                self.assertEqual([(0, 1)], finished)
                self.assertEqual((1, 1), progress[-1][:2])

    def test_splits_large_selections_into_batches(self):
        client = FakeClient()
        entries = [entry(f"/tmp/item_{index:04d}") for index in range(300)]
        worker = DeleteWorker(FakeFactory(client), entries)

        failures, progress, finished = self.run_worker(worker)

        self.assertEqual([], failures)
        self.assertEqual([(300, 0)], finished)
        self.assertEqual(3, len(client.commands))
        self.assertEqual((300, 300), progress[-1][:2])
        self.assertEqual(
            [DELETE_BATCH_ITEMS, DELETE_BATCH_ITEMS, 44],
            [
                command.split(" && if ", 1)[0].count("/tmp/item_")
                for command in client.commands
            ],
        )

    @patch("pgu4_sync.time.sleep")
    def test_batch_failure_falls_back_to_individual_paths(self, _sleep_mock):
        def responder(command):
            remove_command = command.split(" && if ", 1)[0]
            if remove_command.count("/tmp/") > 1:
                return 1, b"batch failed"
            if "/tmp/bad" in remove_command:
                return 1, b"permission denied"
            return 0, b""

        client = FakeClient(responder)
        worker = DeleteWorker(
            FakeFactory(client),
            [entry("/tmp/good"), entry("/tmp/bad")],
        )

        failures, _progress, finished = self.run_worker(worker)

        self.assertEqual(1, len(failures))
        self.assertEqual("/tmp/bad", failures[0][0])
        self.assertIn("permission denied", failures[0][1])
        self.assertIn(f"重试 {DELETE_RETRIES} 次", failures[0][1])
        self.assertEqual([(1, 1)], finished)
        self.assertGreaterEqual(len(client.commands), DELETE_RETRIES + 2)

    @patch("pgu4_sync.time.sleep")
    def test_transient_directory_not_empty_is_retried(self, sleep_mock):
        attempts = {"count": 0}

        def responder(_command):
            attempts["count"] += 1
            if attempts["count"] < 3:
                return 1, b"Directory not empty"
            return 0, b""

        client = FakeClient(responder)
        worker = DeleteWorker(FakeFactory(client), [entry("/tmp/busy", True)])

        failures, progress, finished = self.run_worker(worker)

        self.assertEqual([], failures)
        self.assertEqual([(1, 0)], finished)
        self.assertEqual((1, 1), progress[-1][:2])
        self.assertEqual(3, attempts["count"])
        self.assertEqual(2, sleep_mock.call_count)

    @patch("pgu4_sync.time.sleep")
    def test_persistent_directory_not_empty_has_clear_message(self, _sleep_mock):
        client = FakeClient(lambda _command: (1, b"Directory not empty"))
        worker = DeleteWorker(FakeFactory(client), [entry("/tmp/busy", True)])

        failures, _progress, finished = self.run_worker(worker)

        self.assertEqual([(0, 1)], finished)
        self.assertEqual(1, len(failures))
        self.assertIn("其他程序正在持续写入", failures[0][1])
        self.assertIn(f"重试 {DELETE_RETRIES} 次", failures[0][1])

    def test_connection_failure_reports_every_selected_item(self):
        worker = DeleteWorker(
            FakeFactory(error=TimeoutError("connect timeout")),
            [entry("/tmp/a"), entry("/tmp/b")],
        )

        failures, progress, finished = self.run_worker(worker)

        self.assertEqual({"/tmp/a", "/tmp/b"}, {path for path, _message in failures})
        self.assertEqual((2, 2), progress[-1][:2])
        self.assertEqual([(0, 2)], finished)


if __name__ == "__main__":
    unittest.main()