left4me/l4d2web/tests/test_blueprints.py

import json

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 Blueprint, BlueprintOverlay, Overlay, Server, User


@pytest.fixture
def user_client(tmp_path, monkeypatch):
    db_url = f"sqlite:///{tmp_path/'blueprint.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()
        user_id = user.id
        session.add_all(
            [
                Overlay(name="o1", path="/opt/l4d2/overlays/o1"),
                Overlay(name="o2", path="/opt/l4d2/overlays/o2"),
            ]
        )

    client = app.test_client()
    with client.session_transaction() as sess:
        sess["user_id"] = user_id
        sess["csrf_token"] = "test-token"
    return client


@pytest.fixture
def linked_blueprint(tmp_path, monkeypatch):
    db_url = f"sqlite:///{tmp_path/'linked.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="bob", password_digest=hash_password("secret"), admin=False)
        session.add(user)
        session.flush()

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

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

    client = app.test_client()
    with client.session_transaction() as sess:
        sess["user_id"] = user_id
        sess["csrf_token"] = "test-token"

    return client, blueprint_id


def test_user_can_create_private_blueprint(user_client) -> None:
    payload = {
        "name": "comp",
        "arguments": ["-tickrate 100"],
        "config": ["sv_consistency 1"],
        "overlay_ids": [1, 2],
    }

    response = user_client.post(
        "/blueprints",
        data=json.dumps(payload),
        content_type="application/json",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 201


def _create_other_users_private_overlay() -> int:
    with session_scope() as session:
        other = User(username="mallory", password_digest=hash_password("secret"), admin=False)
        session.add(other)
        session.flush()
        overlay = Overlay(
            name="mallory-private",
            path="mallory-private",
            type="workshop",
            user_id=other.id,
        )
        session.add(overlay)
        session.flush()
        return overlay.id


def test_user_cannot_create_blueprint_with_other_users_private_overlay(user_client) -> None:
    foreign_overlay_id = _create_other_users_private_overlay()
    payload = {
        "name": "bad",
        "arguments": [],
        "config": [],
        "overlay_ids": [foreign_overlay_id],
    }

    response = user_client.post(
        "/blueprints",
        data=json.dumps(payload),
        content_type="application/json",
        headers={"X-CSRF-Token": "test-token"},
    )

    assert response.status_code == 403


def test_user_cannot_update_blueprint_with_other_users_private_overlay(user_client) -> None:
    foreign_overlay_id = _create_other_users_private_overlay()
    create = user_client.post(
        "/blueprints",
        data={"name": "comp", "arguments": "", "config": "", "overlay_ids": ["1"]},
        headers={"X-CSRF-Token": "test-token"},
    )
    assert create.status_code == 302

    response = user_client.post(
        "/blueprints/1",
        data={
            "name": "edited",
            "arguments": "",
            "config": "",
            "overlay_ids": [str(foreign_overlay_id)],
        },
        headers={"X-CSRF-Token": "test-token"},
    )

    assert response.status_code == 403


def test_user_can_create_blueprint_with_system_overlay(user_client) -> None:
    payload = {
        "name": "system-ok",
        "arguments": [],
        "config": [],
        "overlay_ids": [1],
    }

    response = user_client.post(
        "/blueprints",
        data=json.dumps(payload),
        content_type="application/json",
        headers={"X-CSRF-Token": "test-token"},
    )

    assert response.status_code == 201
    blueprint_id = response.get_json()["id"]
    with session_scope() as session:
        link = session.query(BlueprintOverlay).filter_by(blueprint_id=blueprint_id, overlay_id=1).one()
        assert link.position == 0


def test_delete_blueprint_blocked_when_in_use(linked_blueprint) -> None:
    client, blueprint_id = linked_blueprint
    response = client.delete(f"/blueprints/{blueprint_id}", headers={"X-CSRF-Token": "test-token"})
    assert response.status_code == 409


def test_post_delete_blueprint_redirects_to_index(user_client) -> None:
    create = user_client.post(
        "/blueprints",
        data={"name": "doomed", "arguments": "", "config": ""},
        headers={"X-CSRF-Token": "test-token"},
    )
    assert create.status_code == 302

    from sqlalchemy import select

    from l4d2web.models import Blueprint as BlueprintModel

    with session_scope() as session:
        blueprint_id = session.scalars(select(BlueprintModel.id)).one()

    response = user_client.post(
        f"/blueprints/{blueprint_id}/delete",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 302
    assert response.headers["Location"] == "/blueprints"

    with session_scope() as session:
        assert session.scalars(select(BlueprintModel)).all() == []


def test_post_delete_blueprint_blocked_when_in_use(linked_blueprint) -> None:
    client, blueprint_id = linked_blueprint
    response = client.post(
        f"/blueprints/{blueprint_id}/delete",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 409


def test_post_delete_blueprint_returns_404_for_other_user(user_client, tmp_path) -> None:
    with session_scope() as session:
        other = User(username="mallory", password_digest=hash_password("secret"), admin=False)
        session.add(other)
        session.flush()
        foreign = Blueprint(user_id=other.id, name="foreign", arguments="[]", config="[]")
        session.add(foreign)
        session.flush()
        foreign_id = foreign.id

    response = user_client.post(
        f"/blueprints/{foreign_id}/delete",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 404


def test_form_update_preserves_ordered_overlays_and_multiline_fields(user_client) -> None:
    create = user_client.post(
        "/blueprints",
        data={
            "name": "comp",
            "arguments": "-tickrate 100",
            "config": "sv_consistency 1",
            "overlay_ids": ["2", "1"],
            "overlay_position_1": "2",
            "overlay_position_2": "1",
        },
        headers={"X-CSRF-Token": "test-token"},
    )
    assert create.status_code == 302

    update = user_client.post(
        "/blueprints/1",
        data={
            "name": "edited",
            "arguments": "-tickrate 100\n+sv_lan 0",
            "config": "sv_consistency 1\nsv_allow_lobby_connect_only 0",
            "overlay_ids": ["1", "2"],
            "overlay_position_1": "1",
            "overlay_position_2": "2",
        },
        headers={"X-CSRF-Token": "test-token"},
    )

    assert update.status_code == 302
    assert update.headers["Location"] == "/blueprints/1"