1from __future__ import annotations
2
3import os
4from queue import Empty, Queue
5from time import sleep
6
7import pytest
8
9from watchdog.events import (
10    DirCreatedEvent,
11    DirDeletedEvent,
12    DirModifiedEvent,
13    DirMovedEvent,
14    FileCreatedEvent,
15    FileDeletedEvent,
16    FileModifiedEvent,
17    FileMovedEvent,
18)
19from watchdog.observers.api import ObservedWatch
20from watchdog.observers.polling import PollingEmitter as Emitter
21
22from .shell import mkdir, mkdtemp, msize, mv, rm, touch
23
24SLEEP_TIME = 0.4
25TEMP_DIR = mkdtemp()
26
27
28def p(*args):
29    """
30    Convenience function to join the temporary directory path
31    with the provided arguments.
32    """
33    return os.path.join(TEMP_DIR, *args)
34
35
36@pytest.fixture
37def event_queue():
38    return Queue()
39
40
41@pytest.fixture
42def emitter(event_queue):
43    watch = ObservedWatch(TEMP_DIR, recursive=True)
44    em = Emitter(event_queue, watch, timeout=0.2)
45    em.start()
46    yield em
47    em.stop()
48    em.join(5)
49
50
51def test___init__(event_queue, emitter):
52    sleep(SLEEP_TIME)
53    mkdir(p("project"))
54
55    sleep(SLEEP_TIME)
56    mkdir(p("project", "blah"))
57
58    sleep(SLEEP_TIME)
59    touch(p("afile"))
60
61    sleep(SLEEP_TIME)
62    touch(p("fromfile"))
63
64    sleep(SLEEP_TIME)
65    mv(p("fromfile"), p("project", "tofile"))
66
67    sleep(SLEEP_TIME)
68    touch(p("afile"))
69
70    sleep(SLEEP_TIME)
71    mv(p("project", "blah"), p("project", "boo"))
72
73    sleep(SLEEP_TIME)
74    rm(p("project"), recursive=True)
75
76    sleep(SLEEP_TIME)
77    rm(p("afile"))
78
79    sleep(SLEEP_TIME)
80    msize(p("bfile"))
81
82    sleep(SLEEP_TIME)
83    rm(p("bfile"))
84
85    sleep(SLEEP_TIME)
86    emitter.stop()
87
88    # What we need here for the tests to pass is a collection type
89    # that is:
90    #   * unordered
91    #   * non-unique
92    # A multiset! Python's collections.Counter class seems appropriate.
93    expected = {
94        DirModifiedEvent(p()),
95        DirCreatedEvent(p("project")),
96        DirModifiedEvent(p("project")),
97        DirCreatedEvent(p("project", "blah")),
98        FileCreatedEvent(p("afile")),
99        DirModifiedEvent(p()),
100        FileCreatedEvent(p("fromfile")),
101        DirModifiedEvent(p()),
102        DirModifiedEvent(p()),
103        FileModifiedEvent(p("afile")),
104        DirModifiedEvent(p("project")),
105        DirModifiedEvent(p()),
106        FileDeletedEvent(p("project", "tofile")),
107        DirDeletedEvent(p("project", "boo")),
108        DirDeletedEvent(p("project")),
109        DirModifiedEvent(p()),
110        FileDeletedEvent(p("afile")),
111        DirModifiedEvent(p()),
112        FileCreatedEvent(p("bfile")),
113        FileModifiedEvent(p("bfile")),
114        DirModifiedEvent(p()),
115        FileDeletedEvent(p("bfile")),
116    }
117
118    expected.add(FileMovedEvent(p("fromfile"), p("project", "tofile")))
119    expected.add(DirMovedEvent(p("project", "blah"), p("project", "boo")))
120
121    got = set()
122
123    while True:
124        try:
125            event, _ = event_queue.get_nowait()
126            got.add(event)
127        except Empty:
128            break
129
130    assert expected == got
131
132
133def test_delete_watched_dir(event_queue, emitter):
134    rm(p(""), recursive=True)
135
136    sleep(SLEEP_TIME)
137    emitter.stop()
138
139    # What we need here for the tests to pass is a collection type
140    # that is:
141    #   * unordered
142    #   * non-unique
143    # A multiset! Python's collections.Counter class seems appropriate.
144    expected = {
145        DirDeletedEvent(os.path.dirname(p(""))),
146    }
147
148    got = set()
149
150    while True:
151        try:
152            event, _ = event_queue.get_nowait()
153            got.add(event)
154        except Empty:
155            break
156
157    assert expected == got
158