left4me/l4d2web/tests/test_status_and_server_logs.py

import pytest
from datetime import UTC, datetime

from l4d2web.app import create_app
from l4d2web.auth import hash_password
from l4d2web.db import init_db, session_scope
from l4d2web.models import Blueprint, Server, User


@pytest.fixture
def owner_client_with_server(tmp_path, monkeypatch):
    db_url = f"sqlite:///{tmp_path/'status.db'}"
    monkeypatch.setenv("DATABASE_URL", db_url)
    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
    init_db()

    with session_scope() as session:
        user = User(username="alice", password_digest=hash_password("secret"), admin=False)
        session.add(user)
        session.flush()

        blueprint = Blueprint(user_id=user.id, name="default", arguments="[]", config="[]")
        session.add(blueprint)
        session.flush()

        server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)
        session.add(server)
        session.flush()

        user_id = user.id
        server_id = server.id

    client = app.test_client()
    with client.session_transaction() as sess:
        sess["user_id"] = user_id
        sess["pw_changed_at"] = datetime.now(UTC).isoformat()

    return client, server_id


def test_owner_can_stream_server_logs(owner_client_with_server, monkeypatch) -> None:
    client, server_id = owner_client_with_server

    monkeypatch.setattr(
        "l4d2web.services.l4d2_facade.stream_server_logs",
        lambda name, lines=200, follow=True: iter(["first", "second"]),
    )

    response = client.get(f"/servers/{server_id}/logs/stream")
    assert response.status_code == 200


def test_log_stream_translates_heartbeat_to_sse_keepalive(owner_client_with_server, monkeypatch) -> None:
    client, server_id = owner_client_with_server

    monkeypatch.setattr(
        "l4d2web.services.l4d2_facade.stream_server_logs",
        lambda name, lines=200, follow=True: iter(["first", "", "second"]),
    )

    response = client.get(f"/servers/{server_id}/logs/stream")
    assert response.status_code == 200
    body = response.get_data(as_text=True)
    assert "data: first\n\n" in body
    assert "data: second\n\n" in body
    assert ": keepalive\n\n" in body
    assert "data: \n\n" not in body


def test_status_precedence() -> None:
    from l4d2web.services.status import compute_display_state

    assert compute_display_state("start", "stopped") == "starting"


def test_live_state_exposes_recent_overview_and_total_count(owner_client_with_server) -> None:
    """Route must expose recent_players_overview (≤10) and recent_players_total_count.

    The body-text assertions depend on the template change in Task 3.
    The xfail marker will be removed once Task 3 lands.
    """
    from datetime import timedelta
    from l4d2web.models import ServerPlayerSession

    client, server_id = owner_client_with_server

    now = datetime.now(UTC)
    total = 13

    with session_scope() as db:
        for i in range(total):
            db.add(ServerPlayerSession(
                server_id=server_id,
                steam_id_64=str(76561190000000000 + i),
                joined_at=now - timedelta(hours=i + 2),
                left_at=now - timedelta(hours=i + 1),
                name_at_join=f"Player{i}",
                min_ping=10,
                max_ping=50,
            ))

    res = client.get(f"/servers/{server_id}/live-state")
    assert res.status_code == 200
    html = res.get_data(as_text=True)

    # These assertions depend on the Task 3 template changes.
    assert "13 Recent" in html
    assert html.count("recent-chip") <= 10


def test_live_state_recent_header_not_clickable_when_le_10(owner_client_with_server) -> None:
    """When recent player count is ≤ 10, the header must be plain text, not a modal trigger."""
    from datetime import timedelta
    from l4d2web.models import ServerPlayerSession

    client, server_id = owner_client_with_server

    now = datetime.now(UTC)
    total = 4

    with session_scope() as db:
        for i in range(total):
            db.add(ServerPlayerSession(
                server_id=server_id,
                steam_id_64=str(76561190000000000 + i),
                joined_at=now - timedelta(hours=i + 2),
                left_at=now - timedelta(hours=i + 1),
                name_at_join=f"Player{i}",
                min_ping=10,
                max_ping=50,
            ))

    res = client.get(f"/servers/{server_id}/live-state")
    assert res.status_code == 200
    html = res.get_data(as_text=True)

    assert "4 Recent" in html
    assert 'data-inline-modal-open="recent-players-modal"' not in html


def test_live_state_recent_modal_view_returns_single_column_chip_list(owner_client_with_server) -> None:
    """?view=recent-modal renders all recent players (not just the
    10-item overview) in a single-column scrollable list using the
    same chip markup."""
    from datetime import timedelta
    from l4d2web.models import ServerPlayerSession

    client, server_id = owner_client_with_server

    now = datetime.now(UTC)
    total = 13

    with session_scope() as db:
        for i in range(total):
            db.add(ServerPlayerSession(
                server_id=server_id,
                steam_id_64=str(76561190000000000 + i),
                joined_at=now - timedelta(hours=i + 2),
                left_at=now - timedelta(hours=i + 1),
                name_at_join=f"Player{i}",
                min_ping=10,
                max_ping=50,
            ))

    resp = client.get(f"/servers/{server_id}/live-state?view=recent-modal")
    assert resp.status_code == 200
    body = resp.get_data(as_text=True)
    # All 13 chips present (no slice in the modal view).
    assert body.count('class="player-card recent-chip"') == 13
    # Has the modal-list wrapper class.
    assert 'class="player-grid recent recent-modal-list"' in body
    # No "Current" or "N Recent" header — modal view is just the list.
    assert "Recent" not in body or "Recent</button>" not in body  # accept either: the test is about the list rendering, not headers