1from __future__ import annotations
2
3import logging
4import os
5import stat
6import time
7from queue import Empty
8from typing import TYPE_CHECKING
9
10import pytest
11
12from watchdog.events import (
13    DirCreatedEvent,
14    DirDeletedEvent,
15    DirModifiedEvent,
16    DirMovedEvent,
17    FileClosedEvent,
18    FileClosedNoWriteEvent,
19    FileCreatedEvent,
20    FileDeletedEvent,
21    FileModifiedEvent,
22    FileMovedEvent,
23    FileOpenedEvent,
24)
25from watchdog.utils import platform
26
27from .shell import mkdir, mkfile, mv, rm, touch
28
29if TYPE_CHECKING:
30    from .utils import ExpectEvent, P, StartWatching, TestEventQueue
31
32logging.basicConfig(level=logging.DEBUG)
33logger = logging.getLogger(__name__)
34
35
36if platform.is_darwin():
37    # enable more verbose logs
38    fsevents_logger = logging.getLogger("fsevents")
39    fsevents_logger.setLevel(logging.DEBUG)
40
41
42def rerun_filter(exc, *args):
43    time.sleep(5)
44    return bool(issubclass(exc[0], Empty) and platform.is_windows())
45
46
47@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
48def test_create(p: P, event_queue: TestEventQueue, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
49    start_watching()
50    open(p("a"), "a").close()
51
52    expect_event(FileCreatedEvent(p("a")))
53
54    if not platform.is_windows():
55        expect_event(DirModifiedEvent(p()))
56
57    if platform.is_linux():
58        event = event_queue.get(timeout=5)[0]
59        assert event.src_path == p("a")
60        assert isinstance(event, FileOpenedEvent)
61        event = event_queue.get(timeout=5)[0]
62        assert event.src_path == p("a")
63        assert isinstance(event, FileClosedEvent)
64
65
66@pytest.mark.skipif(not platform.is_linux(), reason="FileClosed*Event only supported in GNU/Linux")
67@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
68def test_closed(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
69    with open(p("a"), "a"):
70        start_watching()
71
72    # After file creation/open in append mode
73    event = event_queue.get(timeout=5)[0]
74    assert event.src_path == p("a")
75    assert isinstance(event, FileClosedEvent)
76
77    event = event_queue.get(timeout=5)[0]
78    assert os.path.normpath(event.src_path) == os.path.normpath(p(""))
79    assert isinstance(event, DirModifiedEvent)
80
81    # After read-only, only IN_CLOSE_NOWRITE is emitted
82    open(p("a")).close()
83
84    event = event_queue.get(timeout=5)[0]
85    assert event.src_path == p("a")
86    assert isinstance(event, FileOpenedEvent)
87
88    event = event_queue.get(timeout=5)[0]
89    assert event.src_path == p("a")
90    assert isinstance(event, FileClosedNoWriteEvent)
91
92    assert event_queue.empty()
93
94
95@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
96@pytest.mark.skipif(
97    platform.is_darwin() or platform.is_windows(),
98    reason="Windows and macOS enforce proper encoding",
99)
100def test_create_wrong_encoding(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
101    start_watching()
102    open(p("a_\udce4"), "a").close()
103
104    event = event_queue.get(timeout=5)[0]
105    assert event.src_path == p("a_\udce4")
106    assert isinstance(event, FileCreatedEvent)
107
108    if not platform.is_windows():
109        event = event_queue.get(timeout=5)[0]
110        assert os.path.normpath(event.src_path) == os.path.normpath(p(""))
111        assert isinstance(event, DirModifiedEvent)
112
113
114@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
115def test_delete(p: P, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
116    mkfile(p("a"))
117
118    start_watching()
119    rm(p("a"))
120
121    expect_event(FileDeletedEvent(p("a")))
122
123    if not platform.is_windows():
124        expect_event(DirModifiedEvent(p()))
125
126
127@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
128def test_modify(p: P, event_queue: TestEventQueue, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
129    mkfile(p("a"))
130    start_watching()
131
132    touch(p("a"))
133
134    if platform.is_linux():
135        event = event_queue.get(timeout=5)[0]
136        assert event.src_path == p("a")
137        assert isinstance(event, FileOpenedEvent)
138
139    expect_event(FileModifiedEvent(p("a")))
140
141    if platform.is_linux():
142        event = event_queue.get(timeout=5)[0]
143        assert event.src_path == p("a")
144        assert isinstance(event, FileClosedEvent)
145
146
147@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
148def test_chmod(p: P, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
149    mkfile(p("a"))
150    start_watching()
151
152    # Note: We use S_IREAD here because chmod on Windows only
153    # allows setting the read-only flag.
154    os.chmod(p("a"), stat.S_IREAD)
155
156    expect_event(FileModifiedEvent(p("a")))
157
158    # Reset permissions to allow cleanup.
159    os.chmod(p("a"), stat.S_IWRITE)
160
161
162@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
163def test_move(p: P, event_queue: TestEventQueue, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
164    mkdir(p("dir1"))
165    mkdir(p("dir2"))
166    mkfile(p("dir1", "a"))
167    start_watching()
168
169    mv(p("dir1", "a"), p("dir2", "b"))
170
171    if not platform.is_windows():
172        expect_event(FileMovedEvent(p("dir1", "a"), p("dir2", "b")))
173    else:
174        event = event_queue.get(timeout=5)[0]
175        assert event.src_path == p("dir1", "a")
176        assert isinstance(event, FileDeletedEvent)
177        event = event_queue.get(timeout=5)[0]
178        assert event.src_path == p("dir2", "b")
179        assert isinstance(event, FileCreatedEvent)
180
181    event = event_queue.get(timeout=5)[0]
182    assert event.src_path in [p("dir1"), p("dir2")]
183    assert isinstance(event, DirModifiedEvent)
184
185    if not platform.is_windows():
186        event = event_queue.get(timeout=5)[0]
187        assert event.src_path in [p("dir1"), p("dir2")]
188        assert isinstance(event, DirModifiedEvent)
189
190
191@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
192def test_case_change(
193    p: P,
194    event_queue: TestEventQueue,
195    start_watching: StartWatching,
196    expect_event: ExpectEvent,
197) -> None:
198    mkdir(p("dir1"))
199    mkdir(p("dir2"))
200    mkfile(p("dir1", "file"))
201    start_watching()
202
203    mv(p("dir1", "file"), p("dir2", "FILE"))
204
205    if not platform.is_windows():
206        expect_event(FileMovedEvent(p("dir1", "file"), p("dir2", "FILE")))
207    else:
208        event = event_queue.get(timeout=5)[0]
209        assert event.src_path == p("dir1", "file")
210        assert isinstance(event, FileDeletedEvent)
211        event = event_queue.get(timeout=5)[0]
212        assert event.src_path == p("dir2", "FILE")
213        assert isinstance(event, FileCreatedEvent)
214
215    event = event_queue.get(timeout=5)[0]
216    assert event.src_path in [p("dir1"), p("dir2")]
217    assert isinstance(event, DirModifiedEvent)
218
219    if not platform.is_windows():
220        event = event_queue.get(timeout=5)[0]
221        assert event.src_path in [p("dir1"), p("dir2")]
222        assert isinstance(event, DirModifiedEvent)
223
224
225@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
226def test_move_to(p: P, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
227    mkdir(p("dir1"))
228    mkdir(p("dir2"))
229    mkfile(p("dir1", "a"))
230    start_watching(path=p("dir2"))
231
232    mv(p("dir1", "a"), p("dir2", "b"))
233
234    expect_event(FileCreatedEvent(p("dir2", "b")))
235
236    if not platform.is_windows():
237        expect_event(DirModifiedEvent(p("dir2")))
238
239
240@pytest.mark.skipif(not platform.is_linux(), reason="InotifyFullEmitter only supported in Linux")
241def test_move_to_full(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
242    mkdir(p("dir1"))
243    mkdir(p("dir2"))
244    mkfile(p("dir1", "a"))
245    start_watching(path=p("dir2"), use_full_emitter=True)
246    mv(p("dir1", "a"), p("dir2", "b"))
247
248    event = event_queue.get(timeout=5)[0]
249    assert isinstance(event, FileMovedEvent)
250    assert event.dest_path == p("dir2", "b")
251    assert event.src_path == ""  # Should be blank since the path was not watched
252
253
254@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
255def test_move_from(p: P, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
256    mkdir(p("dir1"))
257    mkdir(p("dir2"))
258    mkfile(p("dir1", "a"))
259    start_watching(path=p("dir1"))
260
261    mv(p("dir1", "a"), p("dir2", "b"))
262
263    expect_event(FileDeletedEvent(p("dir1", "a")))
264
265    if not platform.is_windows():
266        expect_event(DirModifiedEvent(p("dir1")))
267
268
269@pytest.mark.skipif(not platform.is_linux(), reason="InotifyFullEmitter only supported in Linux")
270def test_move_from_full(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
271    mkdir(p("dir1"))
272    mkdir(p("dir2"))
273    mkfile(p("dir1", "a"))
274    start_watching(path=p("dir1"), use_full_emitter=True)
275    mv(p("dir1", "a"), p("dir2", "b"))
276
277    event = event_queue.get(timeout=5)[0]
278    assert isinstance(event, FileMovedEvent)
279    assert event.src_path == p("dir1", "a")
280    assert event.dest_path == ""  # Should be blank since path not watched
281
282
283@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
284def test_separate_consecutive_moves(p: P, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
285    mkdir(p("dir1"))
286    mkfile(p("dir1", "a"))
287    mkfile(p("b"))
288    start_watching(path=p("dir1"))
289    mv(p("dir1", "a"), p("c"))
290    mv(p("b"), p("dir1", "d"))
291
292    dir_modif = DirModifiedEvent(p("dir1"))
293    a_deleted = FileDeletedEvent(p("dir1", "a"))
294    d_created = FileCreatedEvent(p("dir1", "d"))
295
296    expected_events = [a_deleted, dir_modif, d_created, dir_modif]
297
298    if platform.is_windows():
299        expected_events = [a_deleted, d_created]
300
301    if platform.is_bsd():
302        # Due to the way kqueue works, we can't really order
303        # 'Created' and 'Deleted' events in time, so creation queues first
304        expected_events = [d_created, a_deleted, dir_modif, dir_modif]
305
306    for expected_event in expected_events:
307        expect_event(expected_event)
308
309
310@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
311@pytest.mark.skipif(platform.is_bsd(), reason="BSD create another set of events for this test")
312def test_delete_self(p: P, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
313    mkdir(p("dir1"))
314    emitter = start_watching(path=p("dir1"))
315    rm(p("dir1"), recursive=True)
316    expect_event(DirDeletedEvent(p("dir1")))
317    emitter.join(5)
318    assert not emitter.is_alive()
319
320
321@pytest.mark.skipif(
322    platform.is_windows() or platform.is_bsd(),
323    reason="Windows|BSD create another set of events for this test",
324)
325def test_fast_subdirectory_creation_deletion(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
326    root_dir = p("dir1")
327    sub_dir = p("dir1", "subdir1")
328    times = 30
329    mkdir(root_dir)
330    start_watching(path=root_dir)
331    for _ in range(times):
332        mkdir(sub_dir)
333        rm(sub_dir, recursive=True)
334        time.sleep(0.1)  # required for macOS emitter to catch up with us
335    count = {DirCreatedEvent: 0, DirModifiedEvent: 0, DirDeletedEvent: 0}
336    etype_for_dir = {
337        DirCreatedEvent: sub_dir,
338        DirModifiedEvent: root_dir,
339        DirDeletedEvent: sub_dir,
340    }
341    for _ in range(times * 4):
342        event = event_queue.get(timeout=5)[0]
343        logger.debug(event)
344        etype = type(event)
345        count[etype] += 1
346        assert event.src_path == etype_for_dir[etype]
347        assert count[DirCreatedEvent] >= count[DirDeletedEvent]
348        assert count[DirCreatedEvent] + count[DirDeletedEvent] >= count[DirModifiedEvent]
349    assert count == {
350        DirCreatedEvent: times,
351        DirModifiedEvent: times * 2,
352        DirDeletedEvent: times,
353    }
354
355
356@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
357def test_passing_unicode_should_give_unicode(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
358    start_watching(path=str(p()))
359    mkfile(p("a"))
360    event = event_queue.get(timeout=5)[0]
361    assert isinstance(event.src_path, str)
362
363
364@pytest.mark.skipif(
365    platform.is_windows(),
366    reason="Windows ReadDirectoryChangesW supports only" " unicode for paths.",
367)
368def test_passing_bytes_should_give_bytes(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
369    start_watching(path=p().encode())
370    mkfile(p("a"))
371    event = event_queue.get(timeout=5)[0]
372    assert isinstance(event.src_path, bytes)
373
374
375@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
376def test_recursive_on(p: P, event_queue: TestEventQueue, start_watching: StartWatching) -> None:
377    mkdir(p("dir1", "dir2", "dir3"), parents=True)
378    start_watching()
379    touch(p("dir1", "dir2", "dir3", "a"))
380
381    event = event_queue.get(timeout=5)[0]
382    assert event.src_path == p("dir1", "dir2", "dir3", "a")
383    assert isinstance(event, FileCreatedEvent)
384
385    if not platform.is_windows():
386        event = event_queue.get(timeout=5)[0]
387        assert event.src_path == p("dir1", "dir2", "dir3")
388        assert isinstance(event, DirModifiedEvent)
389
390        if platform.is_linux():
391            event = event_queue.get(timeout=5)[0]
392            assert event.src_path == p("dir1", "dir2", "dir3", "a")
393            assert isinstance(event, FileOpenedEvent)
394
395        if not platform.is_bsd():
396            event = event_queue.get(timeout=5)[0]
397            assert event.src_path == p("dir1", "dir2", "dir3", "a")
398            assert isinstance(event, FileModifiedEvent)
399
400
401@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
402def test_recursive_off(
403    p: P,
404    event_queue: TestEventQueue,
405    start_watching: StartWatching,
406    expect_event: ExpectEvent,
407) -> None:
408    mkdir(p("dir1"))
409    start_watching(recursive=False)
410    touch(p("dir1", "a"))
411
412    with pytest.raises(Empty):
413        event_queue.get(timeout=5)
414
415    mkfile(p("b"))
416    expect_event(FileCreatedEvent(p("b")))
417    if not platform.is_windows():
418        expect_event(DirModifiedEvent(p()))
419
420        if platform.is_linux():
421            expect_event(FileOpenedEvent(p("b")))
422            expect_event(FileClosedEvent(p("b")))
423
424    # currently limiting these additional events to macOS only, see https://github.com/gorakhargosh/watchdog/pull/779
425    if platform.is_darwin():
426        mkdir(p("dir1", "dir2"))
427        with pytest.raises(Empty):
428            event_queue.get(timeout=5)
429        mkfile(p("dir1", "dir2", "somefile"))
430        with pytest.raises(Empty):
431            event_queue.get(timeout=5)
432
433        mkdir(p("dir3"))
434        expect_event(DirModifiedEvent(p()))  # the contents of the parent directory changed
435
436        mv(p("dir1", "dir2", "somefile"), p("somefile"))
437        expect_event(FileMovedEvent(p("dir1", "dir2", "somefile"), p("somefile")))
438        expect_event(DirModifiedEvent(p()))
439
440        mv(p("dir1", "dir2"), p("dir2"))
441        expect_event(DirMovedEvent(p("dir1", "dir2"), p("dir2")))
442        expect_event(DirModifiedEvent(p()))
443
444
445@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
446def test_renaming_top_level_directory(
447    p: P,
448    event_queue: TestEventQueue,
449    start_watching: StartWatching,
450    expect_event: ExpectEvent,
451) -> None:
452    start_watching()
453
454    mkdir(p("a"))
455    expect_event(DirCreatedEvent(p("a")))
456    if not platform.is_windows():
457        expect_event(DirModifiedEvent(p()))
458
459    mkdir(p("a", "b"))
460    expect_event(DirCreatedEvent(p("a", "b")))
461    expect_event(DirModifiedEvent(p("a")))
462
463    mv(p("a"), p("a2"))
464    expect_event(DirMovedEvent(p("a"), p("a2")))
465    if not platform.is_windows():
466        expect_event(DirModifiedEvent(p()))
467        expect_event(DirModifiedEvent(p()))
468    expect_event(DirMovedEvent(p("a", "b"), p("a2", "b"), is_synthetic=True))
469
470    if platform.is_bsd():
471        expect_event(DirModifiedEvent(p()))
472
473    open(p("a2", "b", "c"), "a").close()
474
475    # DirModifiedEvent may emitted, but sometimes after waiting time is out.
476    events = []
477    while True:
478        events.append(event_queue.get(timeout=5)[0])
479        if event_queue.empty():
480            break
481
482    assert all(
483        isinstance(e, (FileCreatedEvent, FileMovedEvent, FileOpenedEvent, DirModifiedEvent, FileClosedEvent))
484        for e in events
485    )
486
487    for event in events:
488        if isinstance(event, FileCreatedEvent):
489            assert event.src_path == p("a2", "b", "c")
490        elif isinstance(event, FileMovedEvent):
491            assert event.dest_path == p("a2", "b", "c")
492            assert event.src_path == p("a", "b", "c")
493        elif isinstance(event, DirModifiedEvent):
494            assert event.src_path == p("a2", "b")
495
496
497@pytest.mark.skipif(platform.is_windows(), reason="Windows create another set of events for this test")
498def test_move_nested_subdirectories(
499    p: P,
500    event_queue: TestEventQueue,
501    start_watching: StartWatching,
502    expect_event: ExpectEvent,
503) -> None:
504    mkdir(p("dir1/dir2/dir3"), parents=True)
505    mkfile(p("dir1/dir2/dir3", "a"))
506    start_watching()
507    mv(p("dir1/dir2"), p("dir2"))
508
509    expect_event(DirMovedEvent(p("dir1", "dir2"), p("dir2")))
510    expect_event(DirModifiedEvent(p("dir1")))
511    expect_event(DirModifiedEvent(p()))
512
513    expect_event(DirMovedEvent(p("dir1", "dir2", "dir3"), p("dir2", "dir3"), is_synthetic=True))
514    expect_event(FileMovedEvent(p("dir1", "dir2", "dir3", "a"), p("dir2", "dir3", "a"), is_synthetic=True))
515
516    if platform.is_bsd():
517        event = event_queue.get(timeout=5)[0]
518        assert p(event.src_path) == p()
519        assert isinstance(event, DirModifiedEvent)
520
521        event = event_queue.get(timeout=5)[0]
522        assert p(event.src_path) == p("dir1")
523        assert isinstance(event, DirModifiedEvent)
524
525    touch(p("dir2/dir3", "a"))
526
527    if platform.is_linux():
528        event = event_queue.get(timeout=5)[0]
529        assert event.src_path == p("dir2/dir3", "a")
530        assert isinstance(event, FileOpenedEvent)
531
532    event = event_queue.get(timeout=5)[0]
533    assert event.src_path == p("dir2/dir3", "a")
534    assert isinstance(event, FileModifiedEvent)
535
536
537@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
538@pytest.mark.skipif(
539    not platform.is_windows(),
540    reason="Non-Windows create another set of events for this test",
541)
542def test_move_nested_subdirectories_on_windows(
543    p: P,
544    event_queue: TestEventQueue,
545    start_watching: StartWatching,
546) -> None:
547    mkdir(p("dir1/dir2/dir3"), parents=True)
548    mkfile(p("dir1/dir2/dir3", "a"))
549    start_watching(path=p(""))
550    mv(p("dir1/dir2"), p("dir2"))
551
552    event = event_queue.get(timeout=5)[0]
553    assert event.src_path == p("dir1", "dir2")
554    assert isinstance(event, FileDeletedEvent)
555
556    event = event_queue.get(timeout=5)[0]
557    assert event.src_path == p("dir2")
558    assert isinstance(event, DirCreatedEvent)
559
560    event = event_queue.get(timeout=5)[0]
561    assert event.src_path == p("dir2", "dir3")
562    assert isinstance(event, DirCreatedEvent)
563
564    event = event_queue.get(timeout=5)[0]
565    assert event.src_path == p("dir2", "dir3", "a")
566    assert isinstance(event, FileCreatedEvent)
567
568    touch(p("dir2/dir3", "a"))
569
570    events = []
571    while True:
572        events.append(event_queue.get(timeout=5)[0])
573        if event_queue.empty():
574            break
575
576    assert all(isinstance(e, (FileModifiedEvent, DirModifiedEvent)) for e in events)
577
578    for event in events:
579        if isinstance(event, FileModifiedEvent):
580            assert event.src_path == p("dir2", "dir3", "a")
581        elif isinstance(event, DirModifiedEvent):
582            assert event.src_path in [p("dir2"), p("dir2", "dir3")]
583
584
585@pytest.mark.flaky(max_runs=5, min_passes=1, rerun_filter=rerun_filter)
586@pytest.mark.skipif(platform.is_bsd(), reason="BSD create another set of events for this test")
587def test_file_lifecyle(p: P, start_watching: StartWatching, expect_event: ExpectEvent) -> None:
588    start_watching()
589
590    mkfile(p("a"))
591    touch(p("a"))
592    mv(p("a"), p("b"))
593    rm(p("b"))
594
595    expect_event(FileCreatedEvent(p("a")))
596
597    if not platform.is_windows():
598        expect_event(DirModifiedEvent(p()))
599
600    if platform.is_linux():
601        expect_event(FileOpenedEvent(p("a")))
602        expect_event(FileClosedEvent(p("a")))
603        expect_event(DirModifiedEvent(p()))
604        expect_event(FileOpenedEvent(p("a")))
605
606    expect_event(FileModifiedEvent(p("a")))
607
608    if platform.is_linux():
609        expect_event(FileClosedEvent(p("a")))
610        expect_event(DirModifiedEvent(p()))
611
612    expect_event(FileMovedEvent(p("a"), p("b")))
613
614    if not platform.is_windows():
615        expect_event(DirModifiedEvent(p()))
616        expect_event(DirModifiedEvent(p()))
617
618    expect_event(FileDeletedEvent(p("b")))
619
620    if not platform.is_windows():
621        expect_event(DirModifiedEvent(p()))
622