processkit 3.3.2

Async child-process management for tokio: whole-tree kill-on-drop (no orphans), plus streaming, pipelines, timeouts, and supervision
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

import contextlib
from io import StringIO
from pathlib import Path
import shutil
import tempfile
import unittest
from unittest import mock

import clean_disk


class CleanDiskTests(unittest.TestCase):
    def test_main_continues_after_one_tree_fails(self) -> None:
        with tempfile.TemporaryDirectory() as temporary_directory:
            root = Path(temporary_directory)
            blocked = root / "blocked"
            removable = root / "removable"
            blocked.mkdir()
            removable.mkdir()

            real_rmtree = shutil.rmtree

            def rmtree(path, *args, **kwargs):
                if Path(path) == blocked:
                    raise PermissionError("tree is locked")
                return real_rmtree(path, *args, **kwargs)

            stdout = StringIO()
            stderr = StringIO()
            with (
                mock.patch.object(clean_disk, "target_roots", return_value=[root / "target"]),
                mock.patch.object(
                    clean_disk,
                    "incremental_directories",
                    return_value=[blocked, removable],
                ),
                mock.patch.object(clean_disk.shutil, "rmtree", side_effect=rmtree),
                contextlib.redirect_stdout(stdout),
                contextlib.redirect_stderr(stderr),
            ):
                result = clean_disk.main()

            self.assertEqual(result, 0)
            self.assertTrue(blocked.exists())
            self.assertFalse(removable.exists())
            self.assertIn(f"skip tree after error: {blocked}: tree is locked", stderr.getvalue())
            self.assertIn("clean-disk removed 1 disposable directory", stdout.getvalue())

    def test_readonly_callback_retries_operation_after_chmod(self) -> None:
        for callback in (clean_disk._rmtree_onexc, clean_disk._rmtree_onerror):
            with self.subTest(callback=callback.__name__):
                calls = []

                def remove(path):
                    calls.append(("remove", path))

                with mock.patch.object(clean_disk.os, "chmod") as chmod:
                    callback(remove, "locked.txt", PermissionError("read-only"))

                chmod.assert_called_once_with("locked.txt", clean_disk.stat.S_IWRITE)
                self.assertEqual(calls, [("remove", "locked.txt")])


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