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
208
209
210
211
212
213
214
215
216
|
"""
Music-app enrichment (tag/cover extraction, docs/musicbay.md §2.1/§6) only
ever runs for a group that has an audio_root configured, and only for files
under it — see daemon.py's _enrich_new_audio_entries. Same reasoning as
Videos' video_root gate (test_video_root_gates_enrichment.py), added later:
musicbay.md's original "no root, whole shared tree" call turned out wrong
against a real messy library, where everything under every shared folder
got mixed together with no way to scope Music down to just the actual
music library.
Setting or changing the folder (ops.set_app_directory) fires a one-off sweep
(_enrich_audio_root_now) of whatever it already contains — same shape as
_enrich_video_root_now.
"""
import asyncio
import os
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_node import ops
from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
from meshbay_node.daemon import NodeDaemon
from meshbay_node.indexer import DirectoryIndexer
from meshbay_node.indexer.enrich_audio import AudioEnricher
from meshbay_node.media_cache import MediaCache
from meshbay_node.roster import Roster
from conftest import one_root
pytestmark = pytest.mark.asyncio
# Above indexer.py's MIN_AUDIO_SIZE_BYTES gate — otherwise these fixture
# files would never even be indexed at all, regardless of audio_root.
_AUDIO_BYTES = os.urandom(60 * 1024)
def _free_port() -> int:
import socket
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
async def _make_daemon(tmp_path, shared, group_id):
config = Config(
hub=HubConfig(url="http://localhost:9999", username="testuser"),
node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()),
groups=[GroupConfig(
id=group_id, name="test-group", shared_dir=str(shared),
visibility="private", quic_port=29015,
)],
keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
data_dir=tmp_path / "data",
)
daemon = NodeDaemon(config)
daemon._broadcast_coalesce_secs = 0.01
daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db")
await daemon._media_cache.open()
daemon._audio_enricher = AudioEnricher(daemon._media_cache)
daemon._roster = Roster(db_path=tmp_path / "roster.db")
await daemon._roster.open()
return daemon
async def _teardown(daemon):
await daemon._media_cache.close()
await daemon._roster.close()
async def test_no_audio_root_means_no_enrichment_at_all(tmp_path):
group_id = "a" * 32
shared = tmp_path / "shared"
shared.mkdir()
(shared / "track.mp3").write_bytes(_AUDIO_BYTES)
daemon = await _make_daemon(tmp_path, shared, group_id)
try:
indexer = DirectoryIndexer(
roots=one_root(shared), group_id=group_id,
sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
await indexer.initial_scan()
await daemon._on_index_change(indexer)
await asyncio.sleep(0.05)
assert not daemon._enriched_attempted, (
"a group with no audio_root configured must not enrich anything, "
"not even fall back to the whole index")
finally:
await _teardown(daemon)
async def test_only_entries_under_the_configured_root_are_enriched(tmp_path):
group_id = "a" * 32
shared = tmp_path / "shared"
shared.mkdir()
(shared / "Music").mkdir()
(shared / "Music" / "in-root.mp3").write_bytes(_AUDIO_BYTES)
(shared / "outside.mp3").write_bytes(os.urandom(60 * 1024))
daemon = await _make_daemon(tmp_path, shared, group_id)
try:
await daemon._roster.set_app_directories(group_id, "music", ["shared/Music"], set_by="op")
indexer = DirectoryIndexer(
roots=one_root(shared), group_id=group_id,
sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
await indexer.initial_scan()
await daemon._on_index_change(indexer)
await asyncio.sleep(0.05)
by_name = {e.name: e for e in indexer.index.entries}
assert (group_id, by_name["in-root.mp3"].id) in daemon._enriched_attempted
assert (group_id, by_name["outside.mp3"].id) not in daemon._enriched_attempted, (
"a file outside the configured audio_root must never be enriched")
finally:
await _teardown(daemon)
async def test_setting_the_audio_root_sweeps_what_it_already_contains(tmp_path):
group_id = "a" * 32
shared = tmp_path / "shared"
shared.mkdir()
(shared / "Music").mkdir()
(shared / "Music" / "already-there.mp3").write_bytes(_AUDIO_BYTES)
daemon = await _make_daemon(tmp_path, shared, group_id)
try:
indexer = DirectoryIndexer(
roots=one_root(shared), group_id=group_id,
sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
await indexer.initial_scan()
daemon._state["indexers"][group_id] = indexer
# Broadcast once with nothing configured — nothing should be scheduled.
await daemon._on_index_change(indexer)
await asyncio.sleep(0.05)
assert not daemon._enriched_attempted
# Now the operator points audio_root at the folder that already held
# this file all along.
state = {
"roster": daemon._roster,
# Real roots, because ops now refuses a directory that is not
# inside one — the per-app setters this replaced validated nothing.
"groups_ctx": {group_id: {"roots": one_root(shared)}},
"enrich_app_dirs_fns": {"music": daemon._enrich_audio_root_now},
}
await ops.set_app_directory(state, group_id, "music",
"shared/Music")
await asyncio.sleep(0.05) # let the fire-and-forget sweep actually run
entry = next(iter(indexer.index.entries))
assert (group_id, entry.id) in daemon._enriched_attempted, (
"a file already sitting in the newly-chosen root must be picked "
"up by the sweep, not wait for some unrelated future change")
finally:
await _teardown(daemon)
async def test_the_same_file_shared_into_two_groups_enriches_in_both(tmp_path):
"""
Regression, found live: `entry.id` is a content hash, so the exact same
physical file (byte-for-byte, e.g. a test/demo library reused across
several groups) produces the *same id* wherever it's indexed.
`_enriched_attempted` used to be keyed by that bare id alone — global
across every group this node hosts — so the moment group A's copy got
enriched, group B's otherwise-identical copy read as "already
attempted" and was skipped forever, even though nothing had ever
populated *group B's own* index. Every file in the group manifested as
duration 0 with no artist/album, permanently — nothing else ever
revisits an id once it's in this set. Now keyed by (group_id, id).
"""
shared_a = tmp_path / "shared_a"
shared_a.mkdir()
(shared_a / "Music").mkdir()
(shared_a / "Music" / "track.mp3").write_bytes(_AUDIO_BYTES)
shared_b = tmp_path / "shared_b"
shared_b.mkdir()
(shared_b / "Music").mkdir()
(shared_b / "Music" / "track.mp3").write_bytes(_AUDIO_BYTES) # identical content
group_a, group_b = "a" * 32, "b" * 32
daemon = await _make_daemon(tmp_path, shared_a, group_a)
try:
indexer_a = DirectoryIndexer(
roots=one_root(shared_a), group_id=group_a,
sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
indexer_b = DirectoryIndexer(
roots=one_root(shared_b), group_id=group_b,
sk_node=Ed25519PrivateKey.generate(), gek=generate_gek())
await indexer_a.initial_scan()
await indexer_b.initial_scan()
entry_a = next(iter(indexer_a.index.entries))
entry_b = next(iter(indexer_b.index.entries))
assert entry_a.id == entry_b.id, (
"the fixture itself must produce identical content hashes — "
"otherwise this test isn't exercising the collision at all")
await daemon._roster.set_app_directories(
group_a, "music", ["shared_a/Music"], set_by="op")
await daemon._roster.set_app_directories(
group_b, "music", ["shared_b/Music"], set_by="op")
await daemon._enrich_new_audio_entries(indexer_a, list(indexer_a.index.entries))
await daemon._enrich_new_audio_entries(indexer_b, list(indexer_b.index.entries))
assert (group_a, entry_a.id) in daemon._enriched_attempted
assert (group_b, entry_b.id) in daemon._enriched_attempted, (
"group B's copy must be enriched independently of group A's, "
"even though the two entries share the exact same id")
finally:
await _teardown(daemon)
|