path: root/packages/meshbay-node/tests/test_index_progress.py

"""
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,
        "files_done": 0, "files_total": 0, "kind": "", "root_pos": -1, "queued": 0}


def test_indexing_status_reflects_live_progress():
    progress = IndexProgress(scanning=True, scanned_bytes=500, total_bytes=2000,
                             current_dir="Season 2", root="series", root_pos=1,
                             kind="scan", files_done=3, files_total=9,
                             queued=["archive", "photos"])
    session = _session_with_progress(progress)

    status = session._indexing_status()

    assert status == {"scanning": True, "scanned_bytes": 500, "total_bytes": 2000,
                      "files_done": 3, "files_total": 9, "kind": "scan",
                      "root_pos": 1, "queued": 2}
    assert "current_dir" not in status, \
        "the directory name is operator-local detail, never sent to a member"
    assert not {"series", "archive", "photos"} & {str(v) for v in status.values()}, \
        "a root name reached 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",
    }


# ── /api/index-status (loopback, every group) ───────────────────────────────

def test_every_group_is_described_including_one_still_being_attached():
    """The band reads one route for the whole node. A group in its initial scan
    is in state["indexers"] and not yet in groups_ctx, and must be there."""
    busy, idle = MagicMock(), MagicMock()
    busy.progress = IndexProgress(
        scanning=True, scanned_bytes=10, total_bytes=40, current_dir="2024",
        root="results", root_pos=1, kind="scan", files_done=1, files_total=4,
        queued=["archive"])
    idle.progress = IndexProgress()
    config = MagicMock()
    named = MagicMock()
    named.id, named.name = "a" * 32, "outputs"
    config.groups = [named]
    client = _ui_client({"config": config,
                         "indexers": {"a" * 32: busy, "b" * 32: idle}})

    groups = {g["group_id"]: g for g in client.get("/api/index-status").json()["groups"]}

    assert groups["a" * 32] == {
        "group_id": "a" * 32, "group_name": "outputs", "scanning": True,
        "kind": "scan", "root": "results", "current_dir": "2024",
        "scanned_bytes": 10, "total_bytes": 40, "files_done": 1, "files_total": 4,
        "queued": ["archive"],
    }
    assert groups["b" * 32]["group_name"] == "b" * 8
    assert groups["b" * 32]["scanning"] is False


# ── _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_push_index_progress_carries_counters_never_root_names(tmp_path):
    daemon = _daemon(tmp_path)
    session = MagicMock()
    session._group_id = "a" * 32
    mock_webrtc = MagicMock()
    mock_webrtc._sessions = {"p1": session}
    daemon._webrtc = mock_webrtc

    daemon._push_index_progress("a" * 32, IndexProgress(
        scanning=True, scanned_bytes=10, total_bytes=100, current_dir="2024",
        root="results", root_pos=2, kind="rescan", files_done=5, files_total=50,
        queued=["archive", "photos"]))

    msg = session._send.call_args[0][0]
    assert (msg["kind"], msg["root_pos"], msg["queued"]) == ("rescan", 2, 2)
    assert (msg["files_done"], msg["files_total"]) == (5, 50)
    assert not {"results", "archive", "photos", "2024"} & {str(v) for v in msg.values()}


@pytest.mark.asyncio
async def test_progress_pusher_speaks_while_a_root_only_waits(tmp_path):
    """Between one root's scan ending and the next one taking the lock, nothing
    is scanning — but there is work coming, and the band must not drop it."""
    daemon = _daemon(tmp_path)
    session = MagicMock()
    session._group_id = "a" * 32
    mock_webrtc = MagicMock()
    mock_webrtc._sessions = {"p1": session}
    daemon._webrtc = mock_webrtc

    indexer = MagicMock()
    indexer.group_id = "a" * 32
    indexer.progress = IndexProgress(scanning=False, queued=["archive"])

    task = asyncio.create_task(daemon._progress_pusher(indexer, interval=0.05))
    try:
        await asyncio.sleep(0.12)
        assert session._send.call_count >= 2
        assert session._send.call_args.args[0]["queued"] == 1
    finally:
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass


@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