stackchain-dashboard/tests/test_live_snapshot.py

import asyncio
import json

import pytest

from src import main


@pytest.fixture(autouse=True)
def reset_live_snapshot_task():
    main._live_snapshot_task = None
    yield
    main._live_snapshot_task = None


def payload(response):
    return json.loads(response.body)


@pytest.mark.anyio
async def test_live_snapshot_fetches_user_once_and_updates_work_and_activity(monkeypatch):
    calls = {"user": 0}

    async def user():
        calls["user"] += 1
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def events(authenticated_user):
        assert authenticated_user["login"] == "timmy"
        return [{"type": "push"}]

    async def updates():
        return [{"id": 42, "title": "Mentioned you"}]

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", updates)

    response = await main.live_snapshot()
    result = payload(response)

    assert calls["user"] == 1
    assert result["context"]["user"]["login"] == "timmy"
    assert result["events"] == [{"type": "push"}]
    assert result["notifications"] == [{"id": 42, "title": "Mentioned you"}]
    assert result["sections"] == {
        "context": "fresh", "events": "fresh", "notifications": "fresh"
    }


@pytest.mark.anyio
async def test_live_snapshot_keeps_fresh_context_when_activity_fails(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def failing_events(authenticated_user):
        raise ConnectionError("secret upstream detail")

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", failing_events)
    monkeypatch.setattr(main, "notifications", empty)

    result = payload(await main.live_snapshot())

    assert result["context"]["user"]["login"] == "timmy"
    assert result["events"] is None
    assert result["sections"] == {
        "context": "fresh",
        "events": "temporarily unavailable",
        "notifications": "fresh",
    }
    assert "secret" not in json.dumps(result)


@pytest.mark.anyio
async def test_live_snapshot_keeps_work_and_activity_when_notifications_fail(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def empty():
        return []

    async def events(_authenticated_user):
        return [{"type": "push"}]

    async def failing_updates():
        raise ConnectionError("private notification failure")

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", empty)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", failing_updates)

    result = payload(await main.live_snapshot())

    assert result["context"]["user"]["login"] == "timmy"
    assert result["events"] == [{"type": "push"}]
    assert result["notifications"] is None
    assert result["sections"]["notifications"] == "temporarily unavailable"
    assert "private" not in json.dumps(result)


@pytest.mark.anyio
async def test_live_snapshot_keeps_fresh_activity_when_work_fails(monkeypatch):
    async def user():
        return {"id": 1, "login": "timmy"}

    async def failing_repos():
        raise ConnectionError("work unavailable")

    async def empty():
        return []

    async def events(authenticated_user):
        return [{"type": "push"}]

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", failing_repos)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", empty)

    result = payload(await main.live_snapshot())

    assert result["context"] is None
    assert result["events"] == [{"type": "push"}]
    assert result["sections"] == {
        "context": "temporarily unavailable",
        "events": "fresh",
        "notifications": "fresh",
    }


@pytest.mark.anyio
async def test_live_snapshot_coalesces_only_simultaneous_requests(monkeypatch):
    user_calls = 0
    release = asyncio.Event()

    async def user():
        nonlocal user_calls
        user_calls += 1
        return {"id": 1, "login": "timmy"}

    async def blocked_repos():
        await release.wait()
        return []

    async def empty():
        return []

    async def events(authenticated_user):
        return []

    monkeypatch.setattr(main, "current_user", user)
    monkeypatch.setattr(main, "repos", blocked_repos)
    monkeypatch.setattr(main, "issues", empty)
    monkeypatch.setattr(main, "pull_requests", empty)
    monkeypatch.setattr(main, "activity_events", events)
    monkeypatch.setattr(main, "notifications", empty)

    first = asyncio.create_task(main.live_snapshot())
    await asyncio.sleep(0)
    second = asyncio.create_task(main.live_snapshot())
    await asyncio.sleep(0)
    assert user_calls == 1

    release.set()
    await asyncio.gather(first, second)
    await main.live_snapshot()

    assert user_calls == 2