793 lines
27 KiB
Python
793 lines
27 KiB
Python
import json
|
||
from datetime import UTC, datetime
|
||
|
||
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["pw_changed_at"] = datetime.now(UTC).isoformat()
|
||
sess["csrf_token"] = "test-token"
|
||
|
||
return client, payload
|
||
|
||
|
||
def test_servers_page_without_blueprints_shows_create_blueprint_cta(tmp_path, monkeypatch) -> None:
|
||
db_url = f"sqlite:///{tmp_path/'no_blueprints.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="solo", password_digest=hash_password("secret"), admin=False)
|
||
session.add(user)
|
||
session.flush()
|
||
user_id = user.id
|
||
|
||
client = app.test_client()
|
||
with client.session_transaction() as sess:
|
||
sess["user_id"] = user_id
|
||
sess["pw_changed_at"] = datetime.now(UTC).isoformat()
|
||
sess["csrf_token"] = "test-token"
|
||
|
||
response = client.get("/servers")
|
||
text = response.get_data(as_text=True)
|
||
|
||
assert response.status_code == 200
|
||
assert 'href="/blueprints"' in text
|
||
assert "Create a blueprint first" in text
|
||
assert "disabled" not in text
|
||
|
||
|
||
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", ["", " ", "x" * 129])
|
||
def test_create_server_rejects_invalid_display_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() == []
|
||
|
||
|
||
@pytest.mark.parametrize("name", ["My Practice", "räumlich", "alpha/beta", "..", "Foo"])
|
||
def test_create_server_accepts_free_form_display_names(user_client_with_blueprints, name: str) -> None:
|
||
client, data = user_client_with_blueprints
|
||
response = client.post(
|
||
"/servers",
|
||
data={"name": name, "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert response.status_code == 302
|
||
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Server
|
||
|
||
with session_scope() as session:
|
||
server = session.scalars(select(Server)).one()
|
||
assert server.name == name
|
||
|
||
|
||
def test_create_server_strips_surrounding_whitespace_in_name(user_client_with_blueprints) -> None:
|
||
client, data = user_client_with_blueprints
|
||
response = client.post(
|
||
"/servers",
|
||
data={"name": " spaced ", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert response.status_code == 302
|
||
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Server
|
||
|
||
with session_scope() as session:
|
||
server = session.scalars(select(Server)).one()
|
||
assert server.name == "spaced"
|
||
|
||
|
||
def test_create_server_rejects_duplicate_name_for_same_user(user_client_with_blueprints) -> None:
|
||
client, data = user_client_with_blueprints
|
||
first = client.post(
|
||
"/servers",
|
||
data={"name": "practice", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert first.status_code == 302
|
||
|
||
second = client.post(
|
||
"/servers",
|
||
data={"name": "practice", "port": "27016", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert second.status_code == 409
|
||
assert b"name already in use" in second.data
|
||
|
||
|
||
def test_create_server_allows_same_name_for_different_users(tmp_path, monkeypatch) -> None:
|
||
db_url = f"sqlite:///{tmp_path/'two_users.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:
|
||
alice = User(username="alice", password_digest=hash_password("secret"), admin=False)
|
||
bob = User(username="bob", password_digest=hash_password("secret"), admin=False)
|
||
session.add_all([alice, bob])
|
||
session.flush()
|
||
alice_bp = Blueprint(user_id=alice.id, name="bp", arguments="[]", config="[]")
|
||
bob_bp = Blueprint(user_id=bob.id, name="bp", arguments="[]", config="[]")
|
||
session.add_all([alice_bp, bob_bp])
|
||
session.flush()
|
||
alice_id, bob_id = alice.id, bob.id
|
||
alice_bp_id, bob_bp_id = alice_bp.id, bob_bp.id
|
||
|
||
client = app.test_client()
|
||
|
||
with client.session_transaction() as sess:
|
||
sess["user_id"] = alice_id
|
||
sess["pw_changed_at"] = datetime.now(UTC).isoformat()
|
||
sess["csrf_token"] = "test-token"
|
||
alice_resp = client.post(
|
||
"/servers",
|
||
data={"name": "practice", "port": "27015", "blueprint_id": str(alice_bp_id)},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert alice_resp.status_code == 302
|
||
|
||
with client.session_transaction() as sess:
|
||
sess["user_id"] = bob_id
|
||
sess["pw_changed_at"] = datetime.now(UTC).isoformat()
|
||
sess["csrf_token"] = "test-token"
|
||
bob_resp = client.post(
|
||
"/servers",
|
||
data={"name": "practice", "port": "27016", "blueprint_id": str(bob_bp_id)},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert bob_resp.status_code == 302
|
||
|
||
|
||
def test_create_server_with_empty_port_auto_assigns(user_client_with_blueprints) -> None:
|
||
client, data = user_client_with_blueprints
|
||
response = client.post(
|
||
"/servers",
|
||
data={"name": "alpha", "port": "", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert response.status_code == 302
|
||
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Server
|
||
|
||
with session_scope() as session:
|
||
server = session.scalars(select(Server)).one()
|
||
assert server.port == 27015
|
||
|
||
|
||
def test_create_server_auto_assign_skips_taken_ports(user_client_with_blueprints) -> None:
|
||
client, data = user_client_with_blueprints
|
||
first = client.post(
|
||
"/servers",
|
||
data={"name": "alpha", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert first.status_code == 302
|
||
|
||
second = client.post(
|
||
"/servers",
|
||
data={"name": "beta", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert second.status_code == 302
|
||
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Server
|
||
|
||
with session_scope() as session:
|
||
ports = sorted(session.scalars(select(Server.port)).all())
|
||
assert ports == [27015, 27016]
|
||
|
||
|
||
def test_create_server_returns_409_when_port_range_exhausted(tmp_path, monkeypatch) -> None:
|
||
db_url = f"sqlite:///{tmp_path/'exhausted.db'}"
|
||
monkeypatch.setenv("DATABASE_URL", db_url)
|
||
monkeypatch.setenv("LEFT4ME_PORT_RANGE_START", "30000")
|
||
monkeypatch.setenv("LEFT4ME_PORT_RANGE_END", "30000")
|
||
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="bp", arguments="[]", config="[]")
|
||
session.add(blueprint)
|
||
session.flush()
|
||
user_id = user.id
|
||
blueprint_id = blueprint.id
|
||
|
||
client = app.test_client()
|
||
with client.session_transaction() as sess:
|
||
sess["user_id"] = user_id
|
||
sess["pw_changed_at"] = datetime.now(UTC).isoformat()
|
||
sess["csrf_token"] = "test-token"
|
||
|
||
first = client.post(
|
||
"/servers",
|
||
data={"name": "alpha", "blueprint_id": str(blueprint_id)},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert first.status_code == 302
|
||
|
||
second = client.post(
|
||
"/servers",
|
||
data={"name": "beta", "blueprint_id": str(blueprint_id)},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert second.status_code == 409
|
||
|
||
|
||
def test_delete_operation_redirects_to_index(user_client_with_blueprints) -> None:
|
||
client, data = user_client_with_blueprints
|
||
create = 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.get_json()["id"]
|
||
|
||
response = client.post(
|
||
f"/servers/{server_id}/delete",
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
|
||
assert response.status_code == 302
|
||
assert response.headers["Location"] == "/servers"
|
||
|
||
|
||
def test_servers_page_prefills_blueprint_when_owned(user_client_with_blueprints) -> None:
|
||
client, data = user_client_with_blueprints
|
||
response = client.get(f"/servers?blueprint_id={data['other_blueprint_id']}")
|
||
text = response.get_data(as_text=True)
|
||
|
||
assert response.status_code == 200
|
||
assert f'<option value="{data["other_blueprint_id"]}" selected>' in text
|
||
assert "showModal" in text
|
||
|
||
|
||
def test_servers_page_ignores_foreign_blueprint_id(user_client_with_blueprints) -> None:
|
||
client, _ = user_client_with_blueprints
|
||
|
||
with session_scope() as session:
|
||
other = User(username="bob", 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 = client.get(f"/servers?blueprint_id={foreign_id}")
|
||
text = response.get_data(as_text=True)
|
||
|
||
assert response.status_code == 200
|
||
assert "selected" not in text
|
||
assert "showModal" not in text
|
||
|
||
|
||
def test_servers_page_ignores_non_integer_blueprint_id(user_client_with_blueprints) -> None:
|
||
client, _ = user_client_with_blueprints
|
||
response = client.get("/servers?blueprint_id=abc")
|
||
text = response.get_data(as_text=True)
|
||
|
||
assert response.status_code == 200
|
||
assert "showModal" not in text
|
||
|
||
|
||
def test_servers_page_without_param_does_not_auto_open(user_client_with_blueprints) -> None:
|
||
client, _ = user_client_with_blueprints
|
||
response = client.get("/servers")
|
||
text = response.get_data(as_text=True)
|
||
|
||
assert response.status_code == 200
|
||
assert "showModal" not in text
|
||
|
||
|
||
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}"
|
||
|
||
|
||
def test_create_server_generates_rcon_password(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Server
|
||
|
||
client, data = user_client_with_blueprints
|
||
res = client.post(
|
||
"/servers",
|
||
data={"name": "fresh", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert res.status_code in (200, 201, 302)
|
||
|
||
with session_scope() as db:
|
||
row = db.scalar(select(Server).where(Server.name == "fresh"))
|
||
assert row is not None
|
||
assert len(row.rcon_password) >= 32
|
||
|
||
|
||
def test_servers_index_renders_live_state_badge(user_client_with_blueprints) -> None:
|
||
from datetime import timedelta
|
||
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Server, ServerLiveState
|
||
|
||
client, data = user_client_with_blueprints
|
||
|
||
# Seed one server with a recent snapshot, one without.
|
||
now = datetime.now(UTC)
|
||
with session_scope() as db:
|
||
s_active = Server(
|
||
user_id=data["user_id"],
|
||
blueprint_id=data["blueprint_id"],
|
||
name="active",
|
||
port=27700,
|
||
rcon_password="x",
|
||
actual_state="running",
|
||
)
|
||
s_stale = Server(
|
||
user_id=data["user_id"],
|
||
blueprint_id=data["blueprint_id"],
|
||
name="stale",
|
||
port=27701,
|
||
rcon_password="x",
|
||
actual_state="running",
|
||
)
|
||
db.add_all([s_active, s_stale])
|
||
db.flush()
|
||
db.add(
|
||
ServerLiveState(
|
||
server_id=s_active.id,
|
||
started_at=now,
|
||
last_seen_at=now,
|
||
players=2,
|
||
max_players=4,
|
||
bots=0,
|
||
map="c1m2_streets",
|
||
hibernating=False,
|
||
)
|
||
)
|
||
old = now - timedelta(minutes=5)
|
||
db.add(
|
||
ServerLiveState(
|
||
server_id=s_stale.id,
|
||
started_at=old,
|
||
last_seen_at=old,
|
||
players=0,
|
||
max_players=4,
|
||
bots=0,
|
||
map="c1m1_hotel",
|
||
hibernating=True,
|
||
)
|
||
)
|
||
|
||
res = client.get("/servers")
|
||
html = res.get_data(as_text=True)
|
||
assert "2/4" in html
|
||
assert "c1m2_streets" in html
|
||
# Stale server's map MUST NOT render — fresh badge condition must guard it.
|
||
assert "c1m1_hotel" not in html
|
||
|
||
|
||
def test_live_state_fragment_renders_current_and_recent(user_client_with_blueprints) -> None:
|
||
from datetime import timedelta
|
||
from sqlalchemy import select
|
||
from l4d2web.models import (
|
||
Server, ServerLiveState, ServerPlayerSession, SteamUserProfile,
|
||
)
|
||
|
||
client, data = user_client_with_blueprints
|
||
now = datetime.now(UTC)
|
||
with session_scope() as db:
|
||
srv = Server(
|
||
user_id=data["user_id"], blueprint_id=data["blueprint_id"],
|
||
name="srv", port=27800, rcon_password="x", actual_state="running",
|
||
)
|
||
db.add(srv); db.flush()
|
||
srv_id = srv.id
|
||
db.add(ServerLiveState(
|
||
server_id=srv_id, started_at=now, last_seen_at=now,
|
||
players=1, max_players=4, bots=0, map="c1m2_streets", hibernating=False,
|
||
))
|
||
db.add(ServerPlayerSession(
|
||
server_id=srv_id, steam_id_64="76561197960828710",
|
||
joined_at=now - timedelta(minutes=5), left_at=None,
|
||
name_at_join="Crone", min_ping=40, max_ping=60,
|
||
))
|
||
db.add(ServerPlayerSession(
|
||
server_id=srv_id, steam_id_64="76561198021234567",
|
||
joined_at=now - timedelta(hours=2), left_at=now - timedelta(hours=1),
|
||
name_at_join="OldPlayer", min_ping=20, max_ping=80,
|
||
))
|
||
db.add(SteamUserProfile(
|
||
steam_id_64="76561197960828710",
|
||
persona_name="MrCool42",
|
||
avatar_url="https://avatars.cloudflare.steamstatic.com/cur_medium.jpg",
|
||
fetched_at=now,
|
||
))
|
||
db.add(SteamUserProfile(
|
||
steam_id_64="76561198021234567",
|
||
persona_name="OldPersona",
|
||
avatar_url="https://avatars.cloudflare.steamstatic.com/old_medium.jpg",
|
||
fetched_at=now,
|
||
))
|
||
|
||
res = client.get(f"/servers/{srv_id}/live-state")
|
||
assert res.status_code == 200
|
||
html = res.get_data(as_text=True)
|
||
# Summary
|
||
assert "1/4" in html
|
||
assert "c1m2_streets" in html
|
||
# Current player block
|
||
assert "MrCool42" in html
|
||
assert "cur_medium.jpg" in html
|
||
assert "40-60" in html or "40–60" in html
|
||
# Recent block — only OldPlayer, not MrCool42
|
||
assert "OldPersona" in html
|
||
assert "old_medium.jpg" in html
|
||
# Steam profile links — one for the current player, one for the recent.
|
||
assert "steamcommunity.com/profiles/76561197960828710" in html
|
||
assert "steamcommunity.com/profiles/76561198021234567" in html
|
||
|
||
|
||
def test_reset_operation_enqueues_job_and_stops_desired_state(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Job, Server
|
||
|
||
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"]
|
||
|
||
# Pretend the user already started it.
|
||
with session_scope() as session:
|
||
server = session.scalar(select(Server).where(Server.id == server_id))
|
||
assert server is not None
|
||
server.desired_state = "running"
|
||
|
||
response = client.post(
|
||
f"/servers/{server_id}/reset",
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
|
||
assert response.status_code == 302
|
||
assert response.headers["Location"] == f"/servers/{server_id}"
|
||
|
||
with session_scope() as session:
|
||
server = session.scalar(select(Server).where(Server.id == server_id))
|
||
assert server is not None
|
||
assert server.desired_state == "stopped"
|
||
jobs = session.scalars(
|
||
select(Job).where(Job.server_id == server_id, Job.operation == "reset")
|
||
).all()
|
||
assert len(jobs) == 1
|
||
assert jobs[0].state == "queued"
|
||
|
||
|
||
def test_create_server_hostname_defaults_empty(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
from l4d2web.models import Server
|
||
|
||
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
|
||
|
||
with session_scope() as session:
|
||
server = session.scalar(select(Server).where(Server.name == "alpha"))
|
||
assert server is not None
|
||
assert server.hostname == ""
|
||
|
||
|
||
def test_update_server_hostname_via_form(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
from l4d2web.models import Server
|
||
|
||
client, data = user_client_with_blueprints
|
||
create = client.post(
|
||
"/servers",
|
||
data={"name": "alpha", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
server_id = create.headers["Location"].rsplit("/", 1)[1]
|
||
|
||
update = client.post(
|
||
f"/servers/{server_id}",
|
||
data={"name": "alpha", "hostname": "My Cool Server"},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
assert update.status_code == 302
|
||
|
||
with session_scope() as session:
|
||
server = session.scalar(select(Server).where(Server.name == "alpha"))
|
||
assert server is not None
|
||
assert server.hostname == "My Cool Server"
|
||
|
||
|
||
def test_rename_preserves_hostname(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
from l4d2web.models import Server
|
||
|
||
client, data = user_client_with_blueprints
|
||
create = client.post(
|
||
"/servers",
|
||
data={"name": "alpha", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
server_id = create.headers["Location"].rsplit("/", 1)[1]
|
||
|
||
# Set hostname first
|
||
client.post(
|
||
f"/servers/{server_id}",
|
||
data={"name": "alpha", "hostname": "My Cool Server"},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
|
||
# Rename without sending hostname (simulates rename modal)
|
||
client.post(
|
||
f"/servers/{server_id}",
|
||
data={"name": "beta"},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
|
||
with session_scope() as session:
|
||
server = session.scalar(select(Server).where(Server.name == "beta"))
|
||
assert server is not None
|
||
assert server.hostname == "My Cool Server", "rename must not wipe hostname"
|
||
|
||
|
||
def test_server_detail_no_inline_job_log_pre(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Job, Server
|
||
|
||
client, data = user_client_with_blueprints
|
||
|
||
with session_scope() as db:
|
||
server = Server(
|
||
user_id=data["user_id"],
|
||
blueprint_id=data["blueprint_id"],
|
||
name="jobserver",
|
||
port=27099,
|
||
rcon_password="x",
|
||
actual_state="running",
|
||
)
|
||
db.add(server)
|
||
db.flush()
|
||
job = Job(
|
||
user_id=data["user_id"],
|
||
server_id=server.id,
|
||
operation="start",
|
||
state="running",
|
||
)
|
||
db.add(job)
|
||
db.flush()
|
||
server_id = server.id
|
||
|
||
res = client.get(f"/servers/{server_id}")
|
||
assert res.status_code == 200
|
||
html = res.get_data(as_text=True)
|
||
assert 'class="log-stream job-log"' not in html
|
||
assert 'data-inline-modal-open="job-log-modal"' in html
|
||
|
||
|
||
def test_update_server_clears_hostname(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
from l4d2web.models import Server
|
||
|
||
client, data = user_client_with_blueprints
|
||
create = client.post(
|
||
"/servers",
|
||
data={"name": "alpha", "port": "27015", "blueprint_id": str(data["blueprint_id"])},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
server_id = create.headers["Location"].rsplit("/", 1)[1]
|
||
|
||
# Set hostname first
|
||
client.post(
|
||
f"/servers/{server_id}",
|
||
data={"name": "alpha", "hostname": "My Cool Server"},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
|
||
# Clear it
|
||
client.post(
|
||
f"/servers/{server_id}",
|
||
data={"name": "alpha", "hostname": ""},
|
||
headers={"X-CSRF-Token": "test-token"},
|
||
)
|
||
|
||
with session_scope() as session:
|
||
server = session.scalar(select(Server).where(Server.name == "alpha"))
|
||
assert server is not None
|
||
assert server.hostname == ""
|
||
|
||
|
||
def test_server_detail_renders_state_cluster_and_inspection_strip(user_client_with_blueprints) -> None:
|
||
from sqlalchemy import select
|
||
|
||
from l4d2web.models import Server
|
||
|
||
client, data = user_client_with_blueprints
|
||
|
||
with session_scope() as db:
|
||
server = Server(
|
||
user_id=data["user_id"],
|
||
blueprint_id=data["blueprint_id"],
|
||
name="redesign",
|
||
port=27098,
|
||
rcon_password="x",
|
||
actual_state="stopped",
|
||
)
|
||
db.add(server)
|
||
db.flush()
|
||
server_id = server.id
|
||
|
||
res = client.get(f"/servers/{server_id}")
|
||
assert res.status_code == 200
|
||
html = res.get_data(as_text=True)
|
||
|
||
# State-cluster panel (CSS class order may vary)
|
||
assert 'class="panel state-cluster"' in html or 'class="state-cluster panel"' in html
|
||
|
||
# Inspection strip
|
||
assert "data-tab-strip" in html
|
||
assert 'data-tab="log"' in html
|
||
assert 'data-tab="console"' in html
|
||
assert 'data-tab="files"' in html
|
||
|
||
# Five new modals
|
||
assert 'id="log-modal"' in html
|
||
assert 'id="console-modal"' in html
|
||
assert 'id="files-modal"' in html
|
||
assert 'id="recent-players-modal"' in html
|
||
assert 'id="job-log-modal"' in html
|