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
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
"""
What the operator's progress band is told about the indexing under way.
`progress` said "scanning, this many bytes of that many" and nothing more. That
is one bar with no name on it. A node asked to add a second directory while the
first is still hashing does them one after the other — one scan lock, one
hashing thread — and an operator looking at a bar that jumps back to 0 % cannot
tell a second directory from a scan that started over. So it now says which
root, what kind of walk, how many files, and which roots wait their turn.
"""
import asyncio
import os
from pathlib import Path
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_node.indexer.indexer import DirectoryIndexer
from meshbay_node.roots import RootSet
pytestmark = pytest.mark.asyncio
GROUP = "g" * 32
def _set(*specs) -> RootSet:
return RootSet.build([s if isinstance(s, dict) else {"path": str(s), "name": s.name}
for s in specs])
def _names(idx) -> list[str]:
return sorted(e.name for e in idx.index.entries)
async def _until(predicate, timeout: float = 3.0) -> bool:
deadline = asyncio.get_running_loop().time() + timeout
while not predicate():
if asyncio.get_running_loop().time() > deadline:
return False
await asyncio.sleep(0.02)
return True
class _Held(DirectoryIndexer):
"""Stops before hashing the first file under `hold_under`, and records walks."""
hold_under: Path | None = None
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.gate = asyncio.Event()
self.at_gate = asyncio.Event()
self.walked: list[str] = []
async def _scan_root(self, root, *args, **kwargs):
self.walked.append(root.name)
return await super()._scan_root(root, *args, **kwargs)
async def _hash_or_cached(self, root, file_path):
if self.hold_under is not None and file_path.is_relative_to(self.hold_under):
self.at_gate.set()
await self.gate.wait()
return await super()._hash_or_cached(root, file_path)
def _dirs(tmp_path: Path) -> tuple[Path, Path, Path]:
one, r1, r2 = tmp_path / "one", tmp_path / "r1", tmp_path / "r2"
for d in (one, r1, r2):
d.mkdir()
(one / "a.txt").write_bytes(b"first root")
(r1 / "b.bin").write_bytes(os.urandom(2000))
(r1 / "c.bin").write_bytes(os.urandom(3000))
(r2 / "d.bin").write_bytes(os.urandom(4000))
return one, r1, r2
async def _drain(idx) -> None:
await asyncio.wait_for(asyncio.gather(*list(idx._scan_tasks)), 5)
async def test_the_root_under_way_and_the_ones_waiting_are_named(tmp_path):
one, r1, r2 = _dirs(tmp_path)
idx = _Held(roots=_set(one), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None)
await idx.initial_scan()
idx.walked.clear()
idx.hold_under = r1
try:
await idx.retarget(_set(one, r1), wait=False)
await asyncio.wait_for(idx.at_gate.wait(), 5)
p = idx.progress
assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "scan", "r1", 1)
assert (p.files_done, p.files_total, p.total_bytes) == (0, 2, 5000)
assert p.queued == []
await idx.retarget(_set(one, r1, r2), wait=False)
assert p.queued == ["r2"]
idx.gate.set()
await _drain(idx)
assert idx.walked == ["r1", "r2"], "the second root did not wait for the first"
assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", [])
assert _names(idx) == ["a.txt", "b.bin", "c.bin", "d.bin"]
finally:
idx.gate.set()
await idx.stop()
async def test_a_root_removed_while_it_waited_leaves_the_queue(tmp_path):
one, r1, r2 = _dirs(tmp_path)
idx = _Held(roots=_set(one), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None)
await idx.initial_scan()
idx.walked.clear()
idx.hold_under = r1
try:
await idx.retarget(_set(one, r1), wait=False)
await asyncio.wait_for(idx.at_gate.wait(), 5)
await idx.retarget(_set(one, r1, r2), wait=False)
await idx.retarget(_set(one, r1), wait=False)
assert idx.progress.queued == [], "a removed root is still announced as next"
idx.gate.set()
await _drain(idx)
assert idx.walked == ["r1"]
assert idx.progress.queued == []
finally:
idx.gate.set()
await idx.stop()
async def test_the_initial_scan_names_the_roots_still_to_come(tmp_path):
_, r1, r2 = _dirs(tmp_path)
idx = _Held(roots=_set(r1, r2), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None)
idx.hold_under = r1
scan = asyncio.create_task(idx.initial_scan())
try:
await asyncio.wait_for(idx.at_gate.wait(), 5)
assert (idx.progress.root, idx.progress.queued) == ("r1", ["r2"])
idx.gate.set()
await asyncio.wait_for(scan, 5)
assert idx.progress.queued == []
finally:
idx.gate.set()
await idx.stop()
async def test_a_failed_initial_scan_leaves_nothing_announced(tmp_path):
_, r1, r2 = _dirs(tmp_path)
class _Broken(DirectoryIndexer):
async def _hash_or_cached(self, root, file_path):
raise RuntimeError("simulated failure mid-scan")
idx = _Broken(roots=_set(r1, r2), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None)
with pytest.raises(RuntimeError):
await idx.initial_scan()
p = idx.progress
assert (p.scanning, p.kind, p.root, p.queued) == (False, "", "", [])
async def test_a_plug_waiting_its_turn_is_announced(tmp_path):
one, r1, _ = _dirs(tmp_path)
two = tmp_path / "two"
two.mkdir()
(two / "e.txt").write_bytes(b"removable")
idx = _Held(roots=_set(one, two), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None)
await idx.initial_scan()
idx.eject_root("two")
idx.hold_under = r1
try:
await idx.retarget(
_set(one, {"path": str(two), "name": "two", "ejected": True}, r1), wait=False)
await asyncio.wait_for(idx.at_gate.wait(), 5)
plug = asyncio.create_task(idx.plug_root("two"))
assert await _until(lambda: idx.progress.queued == ["two"])
idx.gate.set()
await asyncio.wait_for(plug, 5)
assert idx.progress.queued == []
assert idx.walked[-1] == "two"
finally:
idx.gate.set()
await idx.stop()
async def test_a_burst_is_described_as_watching(tmp_path):
one, _, _ = _dirs(tmp_path)
idx = DirectoryIndexer(roots=_set(one), group_id=GROUP,
sk_node=Ed25519PrivateKey.generate(), gek=None,
debounce_secs=0.01)
await idx.initial_scan()
idx._loop = asyncio.get_running_loop()
for i in range(2):
f = one / f"new{i}.bin"
f.write_bytes(os.urandom(1000))
idx._schedule_update(f)
await asyncio.sleep(0)
p = idx.progress
assert (p.scanning, p.kind, p.root, p.root_pos) == (True, "watch", "", -1)
assert (p.files_done, p.files_total) == (0, 2)
assert await _until(lambda: not p.scanning)
assert (p.kind, p.files_done, p.files_total) == ("", 2, 2)
|