stackchain-dashboard/tests/test_event_stream.py

import asyncio
import json
from pathlib import Path

import pytest

from src import main
from src import gitea_proxy


DASHBOARD = Path("frontend/index.html")


def test_dashboard_has_realtime_gitea_event_stream_widget():
    html = DASHBOARD.read_text()

    assert "Gitea event stream" in html
    assert "id=\"gitea-events\"" in html
    assert "function paintEventStream" in html
    assert "fetch('api/v1/live'" in html
    assert "paintEventStream(snapshot.events)" in html
    assert "loadEventStream" not in html
    assert "event.actor?.login" in html
    assert "event.repo?.full_name" in html


def test_event_stream_reports_when_activity_was_refreshed():
    html = DASHBOARD.read_text()

    assert 'id="gitea-events-status"' in html
    assert "setEventStreamStatus('Updated ' + fmt(new Date()))" in html


def test_event_stream_reports_refresh_failure_without_erasing_visible_events():
    html = DASHBOARD.read_text()

    assert "setEventStreamStatus('Update failed · showing last activity')" in html
    assert "qs('#gitea-events').innerHTML = '<div class=\"muted\">Event stream unavailable.</div>'" not in html


@pytest.mark.anyio
async def test_event_stream_returns_recent_authenticated_gitea_activity(monkeypatch):
    expected = [{"type": "create", "actor": {"login": "timmy"}}]

    async def fake_activity_events():
        return expected

    monkeypatch.setattr(main, "activity_events", fake_activity_events)

    response = await main.event_stream()

    assert response == expected


@pytest.mark.anyio
async def test_event_stream_returns_retryable_response_when_gitea_exceeds_deadline(monkeypatch):
    cancelled = asyncio.Event()

    async def hanging_activity_events():
        try:
            await asyncio.Event().wait()
        finally:
            cancelled.set()

    monkeypatch.setattr(main, "EVENT_STREAM_TIMEOUT_SECONDS", 0.01)
    monkeypatch.setattr(main, "activity_events", hanging_activity_events)

    response = await main.event_stream()

    assert response.status_code == 503
    assert response.headers["retry-after"] == "1"
    assert json.loads(response.body) == {
        "error": "Gitea event stream request timed out after 0.01s"
    }
    assert cancelled.is_set()


@pytest.mark.anyio
async def test_event_stream_returns_retryable_response_when_gitea_request_fails(monkeypatch):
    async def failing_activity_events():
        raise ConnectionError("connection refused")

    monkeypatch.setattr(main, "activity_events", failing_activity_events)

    response = await main.event_stream()

    assert response.status_code == 503
    assert response.headers["retry-after"] == "5"
    assert json.loads(response.body) == {
        "error": "Gitea event stream is temporarily unavailable"
    }


@pytest.mark.anyio
async def test_activity_events_fetches_feed_for_authenticated_user(monkeypatch):
    paths = []

    async def fake_current_user():
        return {"login": "timmy"}

    async def fake_fetch(path):
        paths.append(path)
        return [{
            "op_type": "push",
            "act_user": {"login": "timmy"},
            "repo": {"full_name": "stackchain/stackchain-dashboard"},
            "created": "2026-08-05T03:00:00Z",
        }]

    monkeypatch.setattr(gitea_proxy, "current_user", fake_current_user)
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)

    result = await gitea_proxy.activity_events()

    assert result == [{
        "type": "push",
        "actor": {"login": "timmy"},
        "repo": {"full_name": "stackchain/stackchain-dashboard"},
        "created_at": "2026-08-05T03:00:00Z",
    }]
    assert paths == ["users/timmy/activities/feeds?limit=20"]


@pytest.mark.anyio
async def test_activity_events_reuses_supplied_authenticated_user(monkeypatch):
    async def unexpected_current_user():
        raise AssertionError("current user must not be fetched twice")

    async def fake_fetch(path):
        assert path == "users/timmy/activities/feeds?limit=20"
        return []

    monkeypatch.setattr(gitea_proxy, "current_user", unexpected_current_user)
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)

    assert await gitea_proxy.activity_events({"login": "timmy"}) == []


@pytest.mark.anyio
async def test_activity_events_skips_malformed_feed_entries(monkeypatch):
    async def fake_current_user():
        return {"login": "timmy"}

    async def fake_fetch(path):
        return [
            None,
            {
                "op_type": "push",
                "act_user": {"login": "timmy"},
                "repo": {"full_name": "stackchain/stackchain-dashboard"},
                "created": "2026-08-06T08:00:00Z",
            },
            "invalid",
        ]

    monkeypatch.setattr(gitea_proxy, "current_user", fake_current_user)
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)

    result = await gitea_proxy.activity_events()

    assert result == [{
        "type": "push",
        "actor": {"login": "timmy"},
        "repo": {"full_name": "stackchain/stackchain-dashboard"},
        "created_at": "2026-08-06T08:00:00Z",
    }]


@pytest.mark.anyio
async def test_activity_events_normalizes_null_feed_payload(monkeypatch):
    async def fake_current_user():
        return {"login": "timmy"}

    async def fake_fetch(path):
        return None

    monkeypatch.setattr(gitea_proxy, "current_user", fake_current_user)
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)

    assert await gitea_proxy.activity_events() == []


@pytest.mark.anyio
async def test_event_stream_rejects_malformed_activity_feed_payload(monkeypatch):
    async def fake_current_user():
        return {"login": "timmy"}

    async def fake_fetch(path):
        return {"message": "unexpected upstream shape"}

    monkeypatch.setattr(gitea_proxy, "current_user", fake_current_user)
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)

    response = await main.event_stream()

    assert getattr(response, "status_code", None) == 503
    assert json.loads(response.body) == {
        "error": "Gitea event stream is temporarily unavailable"
    }


@pytest.mark.anyio
async def test_activity_events_normalizes_malformed_nested_metadata(monkeypatch):
    async def fake_current_user():
        return {"login": "timmy"}

    async def fake_fetch(path):
        return [{
            "op_type": None,
            "act_user": "unknown",
            "repo": None,
            "created": None,
        }]

    monkeypatch.setattr(gitea_proxy, "current_user", fake_current_user)
    monkeypatch.setattr(gitea_proxy, "fetch", fake_fetch)

    assert await gitea_proxy.activity_events() == [{
        "type": "activity",
        "actor": {},
        "repo": {},
        "created_at": "",
    }]