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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
|
import asyncio
from pathlib import Path
from sys import platform
from unittest.mock import patch
import pytest
from aiousbwatcher import AIOUSBWatcher, InotifyNotAvailableError
_INOTIFY_WAIT_TIME = 0.2
@pytest.mark.asyncio
@pytest.mark.skipif(platform == "linux", reason="Inotify is available on this platform")
async def test_aiousbwatcher_not_available() -> None:
with pytest.raises(InotifyNotAvailableError):
watcher = AIOUSBWatcher()
watcher.async_start()
@pytest.mark.asyncio
@pytest.mark.skipif(
platform != "linux", reason="Inotify not available on this platform"
)
async def test_aiousbwatcher_callbacks(tmp_path: Path) -> None:
called: bool = False
def callback() -> None:
nonlocal called
called = True
with patch("aiousbwatcher.impl._PATH", str(tmp_path)):
watcher = AIOUSBWatcher()
unregister = watcher.async_register_callback(callback)
stop = watcher.async_start()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert not called
(tmp_path / "test").touch()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False # type: ignore[unreachable]
unregister()
stop()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert not called
@pytest.mark.asyncio
@pytest.mark.skipif(
platform != "linux", reason="Inotify not available on this platform"
)
async def test_aiousbwatcher_broken_callbacks(
tmp_path: Path, caplog: pytest.LogCaptureFixture
) -> None:
called: bool = False
def callback() -> None:
nonlocal called
called = True
def broken_callback() -> None:
raise Exception("Broken")
with patch("aiousbwatcher.impl._PATH", str(tmp_path)):
watcher = AIOUSBWatcher()
unregister = watcher.async_register_callback(broken_callback)
unregister2 = watcher.async_register_callback(callback)
stop = watcher.async_start()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert not called
(tmp_path / "test").touch()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False # type: ignore[unreachable]
assert "Broken" in caplog.text
unregister()
unregister2()
stop()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert not called
@pytest.mark.asyncio
@pytest.mark.skipif(
platform != "linux", reason="Inotify not available on this platform"
)
async def test_aiousbwatcher_attempt_to_start_twice(tmp_path: Path) -> None:
with patch("aiousbwatcher.impl._PATH", str(tmp_path)):
watcher = AIOUSBWatcher()
stop = watcher.async_start()
with pytest.raises(RuntimeError):
watcher.async_start()
stop()
@pytest.mark.asyncio
@pytest.mark.skipif(
platform != "linux", reason="Inotify not available on this platform"
)
async def test_aiousbwatcher_subdirs_added(tmp_path: Path) -> None:
called: bool = False
def callback() -> None:
nonlocal called
called = True
with patch("aiousbwatcher.impl._PATH", str(tmp_path)):
watcher = AIOUSBWatcher()
unregister = watcher.async_register_callback(callback)
stop = watcher.async_start()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert not called
(tmp_path / "test").mkdir()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False # type: ignore[unreachable]
(tmp_path / "test" / "test2").touch()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False
(tmp_path / "test" / "test2").unlink()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False
(tmp_path / "test").rmdir()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False
(tmp_path / "test").mkdir()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False
(tmp_path / "test" / "test2").touch()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert called
called = False
unregister()
stop()
await asyncio.sleep(_INOTIFY_WAIT_TIME)
assert not called
|
