left4me/l4d2web/tests/test_servers.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, User


@pytest.fixture
def user_client_with_blueprints(tmp_path, monkeypatch):
    db_url = f"sqlite:///{tmp_path/'servers.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_one = Blueprint(user_id=user.id, name="bp1", arguments="[]", config="[]")
        blueprint_two = Blueprint(user_id=user.id, name="bp2", arguments="[]", config="[]")
        session.add_all([blueprint_one, blueprint_two])
        session.flush()
        payload = {
            "user_id": user.id,
            "blueprint_id": blueprint_one.id,
            "other_blueprint_id": blueprint_two.id,
        }

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

    return client, payload


def test_create_server_from_blueprint(user_client_with_blueprints) -> None:
    client, data = user_client_with_blueprints
    payload = {"name": "alpha", "port": 27015, "blueprint_id": data["blueprint_id"]}
    response = client.post(
        "/servers",
        data=json.dumps(payload),
        content_type="application/json",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 201


def test_create_server_from_form_redirects_to_server(user_client_with_blueprints) -> None:
    client, data = user_client_with_blueprints
    response = client.post(
        "/servers",
        data={"name": "alpha", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
        headers={"X-CSRF-Token": "test-token"},
    )

    assert response.status_code == 302
    assert response.headers["Location"].endswith("/servers/1")


def test_reassign_blueprint_anytime(user_client_with_blueprints) -> None:
    client, data = user_client_with_blueprints

    create_payload = {"name": "alpha", "port": 27015, "blueprint_id": data["blueprint_id"]}
    create_response = client.post(
        "/servers",
        data=json.dumps(create_payload),
        content_type="application/json",
        headers={"X-CSRF-Token": "test-token"},
    )
    server_id = create_response.get_json()["id"]

    patch_payload = {"blueprint_id": data["other_blueprint_id"]}
    response = client.patch(
        f"/servers/{server_id}",
        data=json.dumps(patch_payload),
        content_type="application/json",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 200


def test_create_server_duplicate_port(user_client_with_blueprints) -> None:
    client, data = user_client_with_blueprints
    
    # Create the first server
    response = client.post(
        "/servers",
        data={"name": "server-1", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 302
    
    # Try to create a second server with the same port
    response = client.post(
        "/servers",
        data={"name": "server-2", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 409
    assert b"port already in use" in response.data
    
    # Verify the second server was not created by checking how many servers exist
    from sqlalchemy import select
    from l4d2web.db import session_scope
    from l4d2web.models import Server
    
    with session_scope() as session:
        servers = session.scalars(select(Server)).all()
        assert len(servers) == 1
        assert servers[0].name == "server-1"


@pytest.mark.parametrize("bad_name", ["..", "../escape", "foo/bar", " foo", "Foo"])
def test_create_server_rejects_unsafe_names(user_client_with_blueprints, bad_name: str) -> None:
    client, data = user_client_with_blueprints
    response = client.post(
        "/servers",
        data={"name": bad_name, "port": "27015", "blueprint_id": str(data["blueprint_id"])},
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 400

    from sqlalchemy import select

    from l4d2web.models import Server

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


def test_lifecycle_form_creates_queued_job(user_client_with_blueprints) -> None:
    client, data = user_client_with_blueprints
    create_response = client.post(
        "/servers",
        data=json.dumps({"name": "alpha", "port": 27015, "blueprint_id": data["blueprint_id"]}),
        content_type="application/json",
        headers={"X-CSRF-Token": "test-token"},
    )
    server_id = create_response.get_json()["id"]

    response = client.post(
        f"/servers/{server_id}/start",
        headers={"X-CSRF-Token": "test-token"},
    )

    assert response.status_code == 302
    assert response.headers["Location"] == f"/servers/{server_id}"