forked from gorakhargosh/watchdog
-
Notifications
You must be signed in to change notification settings - Fork 0
/
test_watchdog_utils_bricks.py
70 lines (62 loc) · 1.99 KB
/
test_watchdog_utils_bricks.py
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
# -*- coding: utf-8 -*-
import threading
from nose.tools import \
assert_equal, \
assert_true
try:
import queue # IGNORE:F0401
except ImportError:
import Queue as queue # IGNORE:F0401
from watchdog.events import DirModifiedEvent, FileModifiedEvent
from watchdog.utils.bricks import OrderedSetQueue
class TestOrderedSetQueue:
def test_behavior_ordered_set(self):
dir_mod_event = DirModifiedEvent("/path/x")
file_mod_event = FileModifiedEvent('/path/y')
event_list = [
dir_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
file_mod_event,
file_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
file_mod_event,
file_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
dir_mod_event,
file_mod_event,
file_mod_event,
file_mod_event,
file_mod_event,
]
event_set = set(event_list)
event_queue = OrderedSetQueue()
for event in event_list:
event_queue.put(event)
def event_consumer(in_queue):
events = []
while True:
try:
event = in_queue.get(block=True, timeout=0.2)
events.append(event)
in_queue.task_done()
except queue.Empty:
break
# Check set behavior.
assert_true(len(set(events)) == len(events))
assert_equal(set(events), event_set)
# Check order.
assert_equal(events[0], dir_mod_event)
assert_equal(events[1], file_mod_event)
consumer_thread = threading.Thread(target=event_consumer, args=(event_queue,))
consumer_thread.start()
consumer_thread.join()