server-26/drb-c2-core/tests/test_node_sweeper.py

"""
Unit tests for node_sweeper — datetime comparison logic and Firestore update gating.
Firestore and asyncio.to_thread are mocked — no real DB required.
"""
import pytest
import asyncio
from datetime import datetime, timezone, timedelta
from unittest.mock import AsyncMock, patch, MagicMock
from app.internal.node_sweeper import _sweep


def _node(node_id, status, age_seconds):
    """Helper: build a node dict with last_seen age_seconds ago (tz-aware)."""
    return {
        "node_id": node_id,
        "status": status,
        "last_seen": datetime.now(timezone.utc) - timedelta(seconds=age_seconds),
    }


def _node_naive(node_id, status, age_seconds):
    """Helper: tz-naive last_seen (simulates some Firestore SDK versions)."""
    return {
        "node_id": node_id,
        "status": status,
        "last_seen": datetime.utcnow() - timedelta(seconds=age_seconds),
    }


# ---------------------------------------------------------------------------
# Core sweep logic
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_stale_online_node_marked_offline():
    nodes = [_node("node-01", "online", age_seconds=120)]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    mock_fstore.doc_update.assert_called_once_with(
        "nodes", "node-01", {"status": "offline"}
    )


@pytest.mark.asyncio
async def test_stale_recording_node_marked_offline():
    nodes = [_node("node-02", "recording", age_seconds=200)]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    mock_fstore.doc_update.assert_called_once_with(
        "nodes", "node-02", {"status": "offline"}
    )


@pytest.mark.asyncio
async def test_fresh_node_not_touched():
    nodes = [_node("node-03", "online", age_seconds=10)]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    mock_fstore.doc_update.assert_not_called()


@pytest.mark.asyncio
async def test_already_offline_node_skipped():
    """Offline nodes must be skipped even if last_seen is ancient."""
    nodes = [_node("node-04", "offline", age_seconds=9999)]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    mock_fstore.doc_update.assert_not_called()


@pytest.mark.asyncio
async def test_node_with_no_last_seen_skipped():
    nodes = [{"node_id": "node-05", "status": "online", "last_seen": None}]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    mock_fstore.doc_update.assert_not_called()


# ---------------------------------------------------------------------------
# Timezone edge cases
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_tz_naive_last_seen_is_handled():
    """Firestore may return tz-naive datetimes; sweeper must not crash."""
    nodes = [_node_naive("node-06", "online", age_seconds=120)]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    mock_fstore.doc_update.assert_called_once_with(
        "nodes", "node-06", {"status": "offline"}
    )


@pytest.mark.asyncio
async def test_tz_naive_fresh_node_not_touched():
    nodes = [_node_naive("node-07", "online", age_seconds=5)]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    mock_fstore.doc_update.assert_not_called()


# ---------------------------------------------------------------------------
# Batch behaviour
# ---------------------------------------------------------------------------

@pytest.mark.asyncio
async def test_only_stale_nodes_updated_in_batch():
    nodes = [
        _node("node-08", "online", age_seconds=200),   # stale → offline
        _node("node-09", "online", age_seconds=5),     # fresh → skip
        _node("node-10", "offline", age_seconds=500),  # already offline → skip
        _node("node-11", "recording", age_seconds=150), # stale recording → offline
    ]

    with patch("asyncio.to_thread", new=AsyncMock(return_value=nodes)), \
         patch("app.internal.node_sweeper.fstore") as mock_fstore:
        mock_fstore.doc_update = AsyncMock()
        await _sweep()

    assert mock_fstore.doc_update.call_count == 2
    updated_ids = {call.args[1] for call in mock_fstore.doc_update.call_args_list}
    assert updated_ids == {"node-08", "node-11"}