zccache-watcher 1.11.2

PyO3 cdylib hosting the `zccache.watcher._native` extension module
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

import _thread
import tempfile
import threading
import time
import unittest
from pathlib import Path
from unittest import mock

from zccache.watcher import FileWatcher, handle_keyboard_interrupt


class KeyboardInterruptBehaviorTests(unittest.TestCase):
    def test_handle_keyboard_interrupt_reraises_on_main_thread(self) -> None:
        with self.assertRaises(KeyboardInterrupt) as cm:
            handle_keyboard_interrupt(KeyboardInterrupt("stop"))
        self.assertIsInstance(cm.exception.__cause__, KeyboardInterrupt)

    def test_handle_keyboard_interrupt_notifies_main_thread_from_worker(self) -> None:
        called = []
        worker_error = []

        def worker() -> None:
            try:
                handle_keyboard_interrupt(KeyboardInterrupt("worker"))
            except BaseException as exc:  # pragma: no cover - should stay empty
                worker_error.append(exc)

        with mock.patch.object(
            _thread,
            "interrupt_main",
            side_effect=lambda: called.append(True),
        ):
            thread = threading.Thread(target=worker, name="worker-test")
            thread.start()
            thread.join(timeout=2.0)

        self.assertFalse(worker_error)
        self.assertTrue(called)

    def test_poll_predicate_keyboard_interrupt_propagates(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            file_path = root / "watch.cpp"
            file_path.write_text("a\n")

            watcher = FileWatcher(
                root,
                include_globs=["**/*.cpp"],
                debounce_seconds=0.05,
                poll_interval=0.05,
                notification_predicate=lambda *args, **kwargs: (_ for _ in ()).throw(
                    KeyboardInterrupt()
                ),
            )
            try:
                time.sleep(0.2)
                file_path.write_text("b\n")
                with self.assertRaises(KeyboardInterrupt):
                    deadline = time.time() + 2.0
                    while time.time() < deadline:
                        watcher.poll(0.1)
            finally:
                watcher.stop()

    def test_keyboard_interrupt_stop_logs_once_per_watcher(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            file_path = root / "watch.cpp"
            file_path.write_text("a\n")

            watcher = FileWatcher(
                root,
                include_globs=["**/*.cpp"],
                debounce_seconds=0.05,
                poll_interval=0.05,
                notification_predicate=lambda *args, **kwargs: (_ for _ in ()).throw(
                    KeyboardInterrupt()
                ),
            )
            try:
                time.sleep(0.2)
                with mock.patch("builtins.print") as print_mock:
                    file_path.write_text("b\n")
                    with self.assertRaises(KeyboardInterrupt):
                        deadline = time.time() + 2.0
                        while time.time() < deadline:
                            watcher.poll(0.1)
                    stop_logs = [
                        call
                        for call in print_mock.call_args_list
                        if call.args == ("KeyboardInterrupt: watcher stopped",)
                    ]
                    self.assertEqual(len(stop_logs), 1)
            finally:
                watcher.stop()

    def test_callback_keyboard_interrupt_notifies_main_thread(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            file_path = root / "watch.cpp"
            file_path.write_text("a\n")

            called = []

            def callback(_event) -> None:
                raise KeyboardInterrupt()

            watcher = FileWatcher(
                root,
                include_globs=["**/*.cpp"],
                debounce_seconds=0.05,
                poll_interval=0.05,
                callback=callback,
            )
            try:
                time.sleep(0.2)
                with mock.patch.object(_thread, "interrupt_main", side_effect=lambda: called.append(True)):
                    file_path.write_text("b\n")
                    deadline = time.time() + 2.0
                    while time.time() < deadline and not called:
                        time.sleep(0.05)
                self.assertTrue(called)
            finally:
                watcher.stop()