1from __future__ import annotations
2
3import os
4import os.path
5from queue import Empty, Queue
6from time import sleep
7
8import pytest
9
10from watchdog.events import DirCreatedEvent, DirMovedEvent
11from watchdog.observers.api import ObservedWatch
12from watchdog.utils import platform
13
14from .shell import mkdir, mkdtemp, mv, rm
15
16# make pytest aware this is windows only
17if not platform.is_windows():
18    pytest.skip("Windows only.", allow_module_level=True)
19
20from watchdog.observers.read_directory_changes import WindowsApiEmitter
21
22SLEEP_TIME = 2
23
24# Path with non-ASCII
25temp_dir = os.path.join(mkdtemp(), "Strange \N{SNOWMAN}")
26os.makedirs(temp_dir)
27
28
29def p(*args):
30    """
31    Convenience function to join the temporary directory path
32    with the provided arguments.
33    """
34    return os.path.join(temp_dir, *args)
35
36
37@pytest.fixture
38def event_queue():
39    return Queue()
40
41
42@pytest.fixture
43def emitter(event_queue):
44    watch = ObservedWatch(temp_dir, recursive=True)
45    em = WindowsApiEmitter(event_queue, watch, timeout=0.2)
46    yield em
47    em.stop()
48
49
50def test___init__(event_queue, emitter):
51    emitter.start()
52    sleep(SLEEP_TIME)
53    mkdir(p("fromdir"))
54
55    sleep(SLEEP_TIME)
56    mv(p("fromdir"), p("todir"))
57
58    sleep(SLEEP_TIME)
59    emitter.stop()
60
61    # What we need here for the tests to pass is a collection type
62    # that is:
63    #   * unordered
64    #   * non-unique
65    # A multiset! Python's collections.Counter class seems appropriate.
66    expected = {
67        DirCreatedEvent(p("fromdir")),
68        DirMovedEvent(p("fromdir"), p("todir")),
69    }
70
71    got = set()
72
73    while True:
74        try:
75            event, _ = event_queue.get_nowait()
76        except Empty:
77            break
78        else:
79            got.add(event)
80
81    assert expected == got
82
83
84def test_root_deleted(event_queue, emitter):
85    r"""Test the event got when removing the watched folder.
86    The regression to prevent is:
87
88        Exception in thread Thread-1:
89        Traceback (most recent call last):
90        File "watchdog\observers\winapi.py", line 333, in read_directory_changes
91            ctypes.byref(nbytes), None, None)
92        File "watchdog\observers\winapi.py", line 105, in _errcheck_bool
93            raise ctypes.WinError()
94        PermissionError: [WinError 5] Access refused.
95
96        During handling of the above exception, another exception occurred:
97
98        Traceback (most recent call last):
99        File "C:\Python37-32\lib\threading.py", line 926, in _bootstrap_inner
100            self.run()
101        File "watchdog\observers\api.py", line 145, in run
102            self.queue_events(self.timeout)
103        File "watchdog\observers\read_directory_changes.py", line 76, in queue_events
104            winapi_events = self._read_events()
105        File "watchdog\observers\read_directory_changes.py", line 73, in _read_events
106            return read_events(self._whandle, self.watch.path, recursive=self.watch.is_recursive)
107        File "watchdog\observers\winapi.py", line 387, in read_events
108            buf, nbytes = read_directory_changes(handle, path, recursive=recursive)
109        File "watchdog\observers\winapi.py", line 340, in read_directory_changes
110            return _generate_observed_path_deleted_event()
111        File "watchdog\observers\winapi.py", line 298, in _generate_observed_path_deleted_event
112            event = FileNotifyInformation(0, FILE_ACTION_DELETED_SELF, len(path), path.value)
113        TypeError: expected bytes, str found
114    """
115
116    emitter.start()
117    sleep(SLEEP_TIME)
118
119    # This should not fail
120    rm(p(), recursive=True)
121    sleep(SLEEP_TIME)
122
123    # The emitter is automatically stopped, with no error
124    assert not emitter.should_keep_running()
125