6b4eef22c2
Reset stops the systemd service, unmounts the overlay, and rm -rf's both runtime/<name> and instances/<name>, but keeps the Server row, blueprint, and (shared) systemd template. Next Start re-initializes from the current blueprint, so users can clean up logs/caches/accumulated game state without losing the server. Implementation factors a shared _purge_instance helper out of delete_instance; reset_instance reuses it without the existence guard. New "reset" lifecycle op flows through the same route + worker + facade plumbing as the other server ops. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
364 lines
12 KiB
Python
364 lines
12 KiB
Python
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_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["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", ["..", "../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_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["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_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"
|