1from __future__ import annotations
2
3import pytest
4
5from watchdog import events
6from watchdog.utils.bricks import SkipRepeatsQueue
7
8from .markers import cpython_only
9
10
11def basic_actions():
12    q = SkipRepeatsQueue()
13
14    e1 = (2, "fred")
15    e2 = (2, "george")
16    e3 = (4, "sally")
17
18    q.put(e1)
19    q.put(e2)
20    q.put(e3)
21
22    assert e1 == q.get()
23    assert e2 == q.get()
24    assert e3 == q.get()
25    assert q.empty()
26
27
28def test_basic_queue():
29    basic_actions()
30
31
32def test_allow_nonconsecutive():
33    q = SkipRepeatsQueue()
34
35    e1 = (2, "fred")
36    e2 = (2, "george")
37
38    q.put(e1)
39    q.put(e2)
40    q.put(e1)  # repeat the first entry
41
42    assert e1 == q.get()
43    assert e2 == q.get()
44    assert e1 == q.get()
45    assert q.empty()
46
47
48def test_put_with_watchdog_events():
49    # FileSystemEvent.__ne__() uses the key property without
50    # doing any type checking. Since _last_item is set to
51    # None in __init__(), an AttributeError is raised when
52    # FileSystemEvent.__ne__() tries to use None.key
53    queue = SkipRepeatsQueue()
54    dummy_file = "dummy.txt"
55    event = events.FileCreatedEvent(dummy_file)
56    queue.put(event)
57    assert queue.get() is event
58
59
60def test_prevent_consecutive():
61    q = SkipRepeatsQueue()
62
63    e1 = (2, "fred")
64    e2 = (2, "george")
65
66    q.put(e1)
67    q.put(e1)  # repeat the first entry (this shouldn't get added)
68    q.put(e2)
69
70    assert e1 == q.get()
71    assert e2 == q.get()
72    assert q.empty()
73
74
75def test_consecutives_allowed_across_empties():
76    q = SkipRepeatsQueue()
77
78    e1 = (2, "fred")
79
80    q.put(e1)
81    q.put(e1)  # repeat the first entry (this shouldn't get added)
82
83    assert e1 == q.get()
84    assert q.empty()
85
86    q.put(e1)  # this repeat is allowed because 'last' added is now gone from queue
87    assert e1 == q.get()
88    assert q.empty()
89
90
91@cpython_only
92def test_eventlet_monkey_patching():
93    eventlet = pytest.importorskip("eventlet")
94    eventlet.monkey_patch()
95    basic_actions()
96