left4me/l4d2web/tests/test_workshop_routes.py

"""Tests for the workshop overlay routes (add items, remove items, build,
admin refresh)."""
from __future__ import annotations
from datetime import UTC, datetime

from typing import Iterable
from unittest.mock import patch

import pytest

from l4d2web.app import create_app
from l4d2web.auth import hash_password
from l4d2web.db import init_db, session_scope
from l4d2web.models import (
    Job,
    Overlay,
    OverlayWorkshopItem,
    User,
    WorkshopItem,
)
from l4d2web.services import steam_workshop


def _meta(steam_id: str, *, app_id: int = 550, result: int = 1) -> steam_workshop.WorkshopMetadata:
    return steam_workshop.WorkshopMetadata(
        steam_id=steam_id,
        title=f"Item {steam_id}",
        filename=f"{steam_id}.vpk",
        file_url=f"https://example.com/{steam_id}.vpk",
        file_size=42,
        time_updated=1700000000,
        preview_url=f"https://example.com/preview-{steam_id}.jpg",
        consumer_app_id=app_id,
        result=result,
    )


@pytest.fixture
def env_user(tmp_path, monkeypatch):
    db_url = f"sqlite:///{tmp_path/'wr.db'}"
    monkeypatch.setenv("DATABASE_URL", db_url)
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    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("x"), admin=False)
        admin = User(username="admin", password_digest=hash_password("x"), admin=True)
        session.add_all([user, admin])
        session.flush()
        user_id = user.id
        admin_id = admin.id

    def login(uid):
        c = app.test_client()
        with c.session_transaction() as sess:
            sess["user_id"] = uid
            sess["pw_changed_at"] = datetime.now(UTC).isoformat()
            sess["csrf_token"] = "test-token"
        return c

    return app, login, user_id, admin_id


@pytest.fixture
def overlay_for(env_user):
    app, login, user_id, admin_id = env_user

    user_client = login(user_id)
    response = user_client.post(
        "/overlays",
        data={"name": "my-maps", "type": "workshop"},
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 302, response.get_data(as_text=True)

    with session_scope() as session:
        overlay = session.query(Overlay).filter_by(name="my-maps").one()
        overlay_id = overlay.id

    return app, login, user_id, admin_id, overlay_id


def _patch_steam(metas: Iterable[steam_workshop.WorkshopMetadata]):
    return patch.object(steam_workshop, "fetch_metadata_batch", return_value=list(metas))


def test_add_single_item_creates_association_and_enqueues_build(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    with _patch_steam([_meta("1001")]):
        response = user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302
    assert response.headers["Location"].startswith("/jobs/")

    with session_scope() as session:
        n_assoc = session.query(OverlayWorkshopItem).count()
        assert n_assoc == 1
        wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
        assert wi.title == "Item 1001"
        assert wi.preview_url.endswith("preview-1001.jpg")
        # Auto-enqueued build_overlay job.
        jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all()
        assert len(jobs) == 1
        assert jobs[0].state == "queued"
        assert response.headers["Location"] == f"/jobs/{jobs[0].id}"


def test_add_multiline_batch_coalesces_into_one_build_job(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    with _patch_steam([_meta(s) for s in ("1001", "1002", "1003")]):
        response = user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001\n1002\n1003", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302
    assert response.headers["Location"].startswith("/jobs/")

    with session_scope() as session:
        assert session.query(OverlayWorkshopItem).count() == 3
        jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all()
        assert len(jobs) == 1, "multi-item add should coalesce into a single build job"


def test_add_collection_resolves_members(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    with patch.object(steam_workshop, "resolve_collection", return_value=["1001", "1002"]) as resolve:
        with _patch_steam([_meta("1001"), _meta("1002")]):
            response = user_client.post(
                f"/overlays/{overlay_id}/items",
                data={"input": "555", "input_mode": "collection"},
                headers={"X-CSRF-Token": "test-token"},
            )
    assert response.status_code == 302
    assert response.headers["Location"].startswith("/jobs/")
    resolve.assert_called_once_with("555")

    with session_scope() as session:
        assert session.query(OverlayWorkshopItem).count() == 2


def test_add_non_l4d2_item_returns_400(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    def raise_validation(*args, **kwargs):
        raise steam_workshop.WorkshopValidationError("not L4D2")

    with patch.object(steam_workshop, "fetch_metadata_batch", side_effect=raise_validation):
        response = user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "9999", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 400
    assert b"not L4D2" in response.data

    with session_scope() as session:
        assert session.query(WorkshopItem).count() == 0
        assert session.query(OverlayWorkshopItem).count() == 0


def test_add_duplicate_item_does_not_500(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    with _patch_steam([_meta("1001")]):
        first = user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
        assert first.status_code == 302

    with _patch_steam([_meta("1001")]):
        second = user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
        assert second.status_code == 302

    with session_scope() as session:
        assert session.query(OverlayWorkshopItem).count() == 1


def test_remove_item_drops_association_and_enqueues_rebuild(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )

    with session_scope() as session:
        wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
        item_id = wi.id

    response = user_client.post(
        f"/overlays/{overlay_id}/items/{item_id}/delete",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 302
    assert response.headers["Location"].startswith("/jobs/")

    with session_scope() as session:
        assert session.query(OverlayWorkshopItem).count() == 0
        # WorkshopItem itself remains (cache survives the association removal).
        assert session.query(WorkshopItem).filter_by(steam_id="1001").one() is not None
        # Coalesced into the same queued build_overlay job.
        jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all()
        assert len(jobs) == 1


def test_manual_build_button_enqueues_job(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    response = user_client.post(
        f"/overlays/{overlay_id}/build",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 302
    assert response.headers["Location"].startswith("/jobs/")
    with session_scope() as session:
        jobs = session.query(Job).filter_by(operation="build_overlay", overlay_id=overlay_id).all()
        assert len(jobs) == 1


def test_admin_refresh_enqueues_global_job(env_user):
    app, login, user_id, admin_id = env_user
    admin_client = login(admin_id)

    response = admin_client.post(
        "/admin/workshop/refresh",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 302
    assert response.headers["Location"] == "/admin/jobs"
    with session_scope() as session:
        jobs = session.query(Job).filter_by(operation="refresh_workshop_items").all()
        assert len(jobs) == 1
        assert jobs[0].state == "queued"


def test_non_admin_cannot_refresh(env_user):
    app, login, user_id, _admin_id = env_user
    user_client = login(user_id)

    response = user_client.post(
        "/admin/workshop/refresh",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 403


def test_other_user_cannot_modify_workshop_overlay(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for

    with session_scope() as session:
        intruder = User(username="bob", password_digest=hash_password("x"), admin=False)
        session.add(intruder)
        session.flush()
        intruder_id = intruder.id

    intruder_client = login(intruder_id)
    response = intruder_client.post(
        f"/overlays/{overlay_id}/items",
        data={"input": "1001", "input_mode": "items"},
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 403


def test_overlay_refresh_owner_can_refresh_and_enqueues_build(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    with session_scope() as session:
        for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
            job.state = "succeeded"

    fresh_meta = steam_workshop.WorkshopMetadata(
        steam_id="1001", title="Item 1001 (updated)", filename="1001.vpk",
        file_url="https://example.com/1001.vpk", file_size=99, time_updated=1800000000,
        preview_url="https://example.com/preview-1001.jpg", consumer_app_id=550, result=1,
    )
    with patch.object(steam_workshop, "fetch_metadata_batch", return_value=[fresh_meta]) as fetch:
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302
    assert response.headers["Location"].startswith("/jobs/")
    fetch.assert_called_once_with(["1001"], mode="refresh")

    with session_scope() as session:
        wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
        assert wi.title == "Item 1001 (updated)"
        assert wi.time_updated == 1800000000
        new_jobs = session.query(Job).filter_by(
            operation="build_overlay", overlay_id=overlay_id, state="queued"
        ).all()
        assert len(new_jobs) == 1


def test_overlay_refresh_returns_400_when_overlay_empty(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    with patch.object(steam_workshop, "fetch_metadata_batch") as fetch:
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 400
    fetch.assert_not_called()


def test_overlay_refresh_forbidden_for_non_owner(overlay_for, env_user):
    app, login, _user_id, _admin_id, overlay_id = overlay_for
    with session_scope() as session:
        bob = User(username="bob", password_digest=hash_password("x"), admin=False)
        session.add(bob)
        session.flush()
        bob_id = bob.id
    bob_client = login(bob_id)
    response = bob_client.post(
        f"/overlays/{overlay_id}/refresh",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 403


def test_overlay_refresh_admin_can_refresh_anyone(overlay_for):
    app, login, user_id, admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )

    admin_client = login(admin_id)
    with _patch_steam([_meta("1001")]):
        response = admin_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302


def test_overlay_refresh_502_on_steam_error(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    with session_scope() as session:
        for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
            job.state = "succeeded"
        baseline_job_count = session.query(Job).filter_by(
            operation="build_overlay", overlay_id=overlay_id, state="queued"
        ).count()

    with patch.object(steam_workshop, "fetch_metadata_batch", side_effect=Exception("boom")):
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 502
    assert b"steam api error" in response.data

    with session_scope() as session:
        n = session.query(Job).filter_by(
            operation="build_overlay", overlay_id=overlay_id, state="queued"
        ).count()
        assert n == baseline_job_count


def test_overlay_refresh_missing_item_records_last_error(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    with session_scope() as session:
        for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
            job.state = "succeeded"

    with patch.object(steam_workshop, "fetch_metadata_batch", return_value=[]):
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302
    with session_scope() as session:
        wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
        assert "no entry" in wi.last_error