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
|
"""
Indexing status visible node -> client: handshake ack field, the loopback
status route for the Create Group wizard / "add a directory", and the
periodic INDEX_PROGRESS push to already-connected peers. Never the index
itself (see test_daemon.py for that) and never anything sent to the hub.
"""
import asyncio
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from unittest.mock import MagicMock
from fastapi.testclient import TestClient
from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig
from meshbay_node.daemon import NodeDaemon
from meshbay_node.indexer.indexer import IndexProgress
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from meshbay_node.ui.app import create_ui_app
def _session_with_progress(progress: IndexProgress | None, group_id: str = "g" * 32):
index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate())
session = WebRTCPeerSession.__new__(WebRTCPeerSession)
group_ctx = {"index": index}
if progress is not None:
group_ctx["progress"] = progress
session._ctx = {"groups": {group_id: group_ctx}}
session._group_id = group_id
return session
# ── _indexing_status() ───────────────────────────────────────────────────────
def test_indexing_status_defaults_idle_when_no_progress_tracked():
session = _session_with_progress(None)
assert session._indexing_status() == {
"scanning": False, "scanned_bytes": 0, "total_bytes": 0}
def test_indexing_status_reflects_live_progress():
progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000,
current_dir="StarWars")
session = _session_with_progress(progress)
status = session._indexing_status()
assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000}
assert "current_dir" not in status, \
"the directory name is operator-local detail, never sent to a member"
# ── /api/groups/{id}/index-status (loopback) ────────────────────────────────
def _ui_client(state: dict) -> TestClient:
return TestClient(create_ui_app({"status": "running", "groups_ctx": {},
"indexes": {}, **state}))
def test_index_status_route_idle_for_unknown_group():
client = _ui_client({"indexers": {}})
resp = client.get("/api/groups/unknown-group/index-status")
assert resp.status_code == 200
assert resp.json() == {"scanning": False, "scanned_bytes": 0,
"total_bytes": 0, "current_dir": ""}
def test_index_status_route_reflects_indexer_progress():
fake_indexer = MagicMock()
fake_indexer.progress = IndexProgress(
scanning=True, scanned_bytes=1_000_000, total_bytes=4_000_000_000,
current_dir="2024")
client = _ui_client({"indexers": {"g" * 32: fake_indexer}})
resp = client.get(f"/api/groups/{'g' * 32}/index-status")
assert resp.json() == {
"scanning": True, "scanned_bytes": 1_000_000,
"total_bytes": 4_000_000_000, "current_dir": "2024",
}
# ── _push_index_progress / _progress_pusher ─────────────────────────────────
def _daemon(tmp_path) -> NodeDaemon:
config = Config(
hub=HubConfig(url="http://localhost:9999", username="testuser"),
node=NodeConfig(),
groups=[],
keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
data_dir=tmp_path / "data",
)
return NodeDaemon(config)
@pytest.mark.asyncio
async def test_push_index_progress_only_reaches_same_group_peers(tmp_path):
daemon = _daemon(tmp_path)
same_group = MagicMock()
same_group._group_id = "a" * 32
same_group._send = MagicMock()
other_group = MagicMock()
other_group._group_id = "b" * 32
other_group._send = MagicMock()
mock_webrtc = MagicMock()
mock_webrtc._sessions = {"p1": same_group, "p2": other_group}
daemon._webrtc = mock_webrtc
progress = IndexProgress(scanning=True, scanned_bytes=10, total_bytes=100)
daemon._push_index_progress("a" * 32, progress)
same_group._send.assert_called_once()
msg = same_group._send.call_args[0][0]
assert msg["type"] == "index_progress"
assert msg["group_id"] == "a" * 32
assert msg["scanning"] is True
assert msg["scanned_bytes"] == 10
assert msg["total_bytes"] == 100
other_group._send.assert_not_called()
@pytest.mark.asyncio
async def test_progress_pusher_pushes_while_scanning_then_one_final_push(tmp_path):
daemon = _daemon(tmp_path)
session = MagicMock()
session._group_id = "a" * 32
session._send = MagicMock()
mock_webrtc = MagicMock()
mock_webrtc._sessions = {"p1": session}
daemon._webrtc = mock_webrtc
indexer = MagicMock()
indexer.group_id = "a" * 32
indexer.progress = IndexProgress(scanning=True, scanned_bytes=0, total_bytes=100)
task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05))
try:
# Two ticks while still scanning.
await asyncio.sleep(0.12)
assert session._send.call_count >= 2
assert all(c.args[0]["scanning"] is True for c in session._send.call_args_list)
# Scan finishes between ticks.
indexer.progress.scanning = False
calls_before = session._send.call_count
await asyncio.sleep(0.07)
assert session._send.call_count == calls_before + 1, \
"exactly one final push must follow the False transition"
assert session._send.call_args.args[0]["scanning"] is False
# Nothing further once idle.
calls_after_final = session._send.call_count
await asyncio.sleep(0.15)
assert session._send.call_count == calls_after_final, \
"no more pushes once idle and already reported"
finally:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
|