92d6ebbe82
Adds two managed system overlays (l4d2center-maps, cedapug-maps) that fetch curated map archives from upstream sources and reconcile addons symlinks for non-Steam maps. A daily systemd timer enqueues a coalesced refresh_global_overlays worker job; downloads, extraction, and rebuilds run in the existing job worker and surface in the job log UI. Schema: GlobalOverlaySource / GlobalOverlayItem / GlobalOverlayItemFile plus nullable Job.user_id so system jobs render as "system" in the UI. The new builder reconciles symlinks against the per-source vpk cache and leaves foreign symlinks untouched. Initialize-time guard refuses to mount a partial overlay if any expected vpk is missing from cache. Refresh service uses shutil.move to handle EXDEV when /tmp and the cache live on different filesystems. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
457 lines
16 KiB
Python
457 lines
16 KiB
Python
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, GlobalOverlaySource, Overlay, User
|
|
from l4d2web.services.security import validate_overlay_ref
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_client(tmp_path, monkeypatch):
|
|
db_url = f"sqlite:///{tmp_path/'admin_overlay.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:
|
|
admin = User(username="admin", password_digest=hash_password("secret"), admin=True)
|
|
session.add(admin)
|
|
session.flush()
|
|
admin_id = admin.id
|
|
|
|
client = app.test_client()
|
|
with client.session_transaction() as sess:
|
|
sess["user_id"] = admin_id
|
|
sess["csrf_token"] = "test-token"
|
|
return client
|
|
|
|
|
|
@pytest.fixture
|
|
def user_client_with_overlay(tmp_path, monkeypatch):
|
|
db_url = f"sqlite:///{tmp_path/'user_overlay.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("secret"), admin=False)
|
|
session.add(user)
|
|
# System external overlay (no user_id), pre-existing.
|
|
session.add(Overlay(name="standard", path="standard", type="external", user_id=None))
|
|
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"
|
|
return client
|
|
|
|
|
|
def test_user_can_view_overlay_catalog(user_client_with_overlay) -> None:
|
|
response = user_client_with_overlay.get("/overlays")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "standard" in text
|
|
# Non-admin users can create workshop overlays, so the Create button shows.
|
|
assert "Create overlay" in text
|
|
|
|
|
|
def test_non_admin_can_view_managed_global_system_overlay(user_client_with_overlay) -> None:
|
|
_create_managed_global_overlay()
|
|
|
|
response = user_client_with_overlay.get("/overlays")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "l4d2center-maps" in text
|
|
|
|
|
|
def test_admin_can_view_overlay_edit_controls(admin_client) -> None:
|
|
response = admin_client.get("/overlays")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "Create overlay" in text
|
|
assert 'action="/overlays"' in text
|
|
|
|
|
|
def test_admin_can_create_external_overlay(admin_client) -> None:
|
|
response = admin_client.post(
|
|
"/overlays",
|
|
data={"name": "standard", "type": "external"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 302
|
|
# Redirect to the new detail page now that paths are auto-generated.
|
|
assert response.headers["Location"].startswith("/overlays/")
|
|
|
|
|
|
def test_admin_cannot_create_managed_global_overlay_type(admin_client) -> None:
|
|
response = admin_client.post(
|
|
"/overlays",
|
|
data={"name": "managed", "type": "l4d2center_maps"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "unknown overlay type" in response.get_data(as_text=True)
|
|
|
|
|
|
@pytest.mark.parametrize("overlay_ref", [" standard", "standard ", "a//b", "a/", "./a", "a/.", "."])
|
|
def test_overlay_ref_rejects_unsafe_components(overlay_ref: str) -> None:
|
|
with pytest.raises(ValueError):
|
|
validate_overlay_ref(overlay_ref)
|
|
|
|
|
|
def test_non_admin_cannot_create_external_overlay(user_client_with_overlay) -> None:
|
|
response = user_client_with_overlay.post(
|
|
"/overlays",
|
|
data={"name": "bad", "type": "external"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_user_can_create_workshop_overlay(user_client_with_overlay) -> None:
|
|
response = user_client_with_overlay.post(
|
|
"/overlays",
|
|
data={"name": "my-maps", "type": "workshop"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 302
|
|
with session_scope() as session:
|
|
overlay = session.query(Overlay).filter_by(name="my-maps").one()
|
|
assert overlay.type == "workshop"
|
|
assert overlay.user_id is not None
|
|
assert overlay.path == str(overlay.id)
|
|
|
|
|
|
def test_workshop_overlay_directory_is_created_on_disk(user_client_with_overlay, tmp_path) -> None:
|
|
response = user_client_with_overlay.post(
|
|
"/overlays",
|
|
data={"name": "my-maps", "type": "workshop"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 302
|
|
with session_scope() as session:
|
|
overlay = session.query(Overlay).filter_by(name="my-maps").one()
|
|
overlay_id = overlay.id
|
|
assert (tmp_path / "overlays" / str(overlay_id)).is_dir()
|
|
|
|
|
|
def test_two_users_can_have_workshop_overlay_with_same_name(tmp_path, monkeypatch) -> None:
|
|
# Set up a fresh app with two users.
|
|
db_url = f"sqlite:///{tmp_path/'shared.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:
|
|
for username in ("alice", "bob"):
|
|
session.add(User(username=username, password_digest=hash_password("x"), admin=False))
|
|
session.flush()
|
|
alice_id, bob_id = (
|
|
session.query(User).filter_by(username="alice").one().id,
|
|
session.query(User).filter_by(username="bob").one().id,
|
|
)
|
|
|
|
def client_for(uid):
|
|
c = app.test_client()
|
|
with c.session_transaction() as sess:
|
|
sess["user_id"] = uid
|
|
sess["csrf_token"] = "test-token"
|
|
return c
|
|
|
|
for uid in (alice_id, bob_id):
|
|
r = client_for(uid).post(
|
|
"/overlays",
|
|
data={"name": "my-maps", "type": "workshop"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert r.status_code == 302
|
|
|
|
with session_scope() as session:
|
|
rows = session.query(Overlay).filter_by(name="my-maps").all()
|
|
assert {r.user_id for r in rows} == {alice_id, bob_id}
|
|
|
|
|
|
def test_admin_can_update_and_delete_overlay(admin_client) -> None:
|
|
create = admin_client.post(
|
|
"/overlays",
|
|
data={"name": "standard", "type": "external"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert create.status_code == 302
|
|
|
|
with session_scope() as session:
|
|
overlay_id = session.query(Overlay).filter_by(name="standard").one().id
|
|
|
|
update = admin_client.post(
|
|
f"/overlays/{overlay_id}",
|
|
data={"name": "edited"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert update.status_code == 302
|
|
|
|
delete = admin_client.post(
|
|
f"/overlays/{overlay_id}/delete",
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert delete.status_code == 302
|
|
|
|
|
|
def _create_managed_global_overlay() -> int:
|
|
with session_scope() as session:
|
|
overlay = Overlay(
|
|
name="l4d2center-maps",
|
|
path="managed-l4d2center",
|
|
type="l4d2center_maps",
|
|
user_id=None,
|
|
)
|
|
session.add(overlay)
|
|
session.flush()
|
|
session.add(
|
|
GlobalOverlaySource(
|
|
overlay_id=overlay.id,
|
|
source_key="l4d2center-maps",
|
|
source_type="l4d2center_csv",
|
|
source_url="https://l4d2center.com/maps/servers/index.csv",
|
|
)
|
|
)
|
|
return overlay.id
|
|
|
|
|
|
def test_admin_cannot_update_managed_global_overlay(admin_client) -> None:
|
|
overlay_id = _create_managed_global_overlay()
|
|
|
|
response = admin_client.post(
|
|
f"/overlays/{overlay_id}",
|
|
data={"name": "renamed"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_admin_cannot_delete_managed_global_overlay(admin_client) -> None:
|
|
overlay_id = _create_managed_global_overlay()
|
|
|
|
response = admin_client.post(
|
|
f"/overlays/{overlay_id}/delete",
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_admin_overlay_detail_hides_edit_for_managed_global_overlay(admin_client) -> None:
|
|
overlay_id = _create_managed_global_overlay()
|
|
|
|
response = admin_client.get(f"/overlays/{overlay_id}")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert f'action="/overlays/{overlay_id}"' not in text
|
|
assert "delete-overlay-modal" not in text
|
|
|
|
|
|
|
|
def test_update_overlay_rejects_duplicate_name(admin_client) -> None:
|
|
ids: list[int] = []
|
|
for name in ("standard", "competitive"):
|
|
response = admin_client.post(
|
|
"/overlays",
|
|
data={"name": name, "type": "external"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 302
|
|
with session_scope() as session:
|
|
ids = [
|
|
session.query(Overlay).filter_by(name="standard").one().id,
|
|
session.query(Overlay).filter_by(name="competitive").one().id,
|
|
]
|
|
|
|
response = admin_client.post(
|
|
f"/overlays/{ids[1]}",
|
|
data={"name": "standard"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_overlay_detail_page_lists_using_blueprints(admin_client) -> None:
|
|
create = admin_client.post(
|
|
"/overlays",
|
|
data={"name": "shared", "type": "external"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert create.status_code == 302
|
|
with session_scope() as session:
|
|
overlay_id = session.query(Overlay).filter_by(name="shared").one().id
|
|
|
|
with session_scope() as session:
|
|
admin = session.query(User).filter_by(username="admin").one()
|
|
bp_one = Blueprint(user_id=admin.id, name="alpha-bp", arguments="[]", config="[]")
|
|
bp_two = Blueprint(user_id=admin.id, name="beta-bp", arguments="[]", config="[]")
|
|
session.add_all([bp_one, bp_two])
|
|
session.flush()
|
|
session.add(BlueprintOverlay(blueprint_id=bp_one.id, overlay_id=overlay_id, position=0))
|
|
session.add(BlueprintOverlay(blueprint_id=bp_two.id, overlay_id=overlay_id, position=0))
|
|
|
|
response = admin_client.get(f"/overlays/{overlay_id}")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "alpha-bp" in text
|
|
assert "beta-bp" in text
|
|
assert "Used by" in text
|
|
|
|
|
|
def test_non_admin_overlay_detail_only_lists_own_using_blueprints(user_client_with_overlay) -> None:
|
|
overlay_id = _create_managed_global_overlay()
|
|
with session_scope() as session:
|
|
alice = session.query(User).filter_by(username="alice").one()
|
|
other = User(username="mallory", password_digest=hash_password("secret"), admin=False)
|
|
session.add(other)
|
|
session.flush()
|
|
|
|
own_bp = Blueprint(user_id=alice.id, name="own-bp", arguments="[]", config="[]")
|
|
other_bp = Blueprint(user_id=other.id, name="other-private-bp", arguments="[]", config="[]")
|
|
session.add_all([own_bp, other_bp])
|
|
session.flush()
|
|
session.add(BlueprintOverlay(blueprint_id=own_bp.id, overlay_id=overlay_id, position=0))
|
|
session.add(BlueprintOverlay(blueprint_id=other_bp.id, overlay_id=overlay_id, position=0))
|
|
|
|
response = user_client_with_overlay.get(f"/overlays/{overlay_id}")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "own-bp" in text
|
|
assert "other-private-bp" not in text
|
|
|
|
|
|
def test_blueprint_edit_lists_system_and_owned_overlays_only(user_client_with_overlay) -> None:
|
|
system_overlay_id = _create_managed_global_overlay()
|
|
with session_scope() as session:
|
|
alice = session.query(User).filter_by(username="alice").one()
|
|
other = User(username="mallory", password_digest=hash_password("secret"), admin=False)
|
|
session.add(other)
|
|
session.flush()
|
|
foreign_overlay = Overlay(
|
|
name="other-private-workshop",
|
|
path="other-private-workshop",
|
|
type="workshop",
|
|
user_id=other.id,
|
|
)
|
|
blueprint = Blueprint(user_id=alice.id, name="alice-bp", arguments="[]", config="[]")
|
|
session.add_all([foreign_overlay, blueprint])
|
|
session.flush()
|
|
blueprint_id = blueprint.id
|
|
|
|
response = user_client_with_overlay.get(f"/blueprints/{blueprint_id}")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "l4d2center-maps" in text
|
|
assert f'value="{system_overlay_id}"' in text
|
|
assert "other-private-workshop" not in text
|
|
|
|
|
|
def test_overlay_detail_page_404_when_missing(admin_client) -> None:
|
|
response = admin_client.get("/overlays/999")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_overlay_detail_hides_edit_for_non_admin_external(user_client_with_overlay) -> None:
|
|
# The seeded "standard" external overlay (id=1, user_id=NULL) is admin-only edit.
|
|
response = user_client_with_overlay.get("/overlays/1")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "standard" in text
|
|
assert 'action="/overlays/1"' not in text
|
|
assert "delete-overlay-modal" not in text
|
|
|
|
|
|
def test_non_admin_cannot_view_other_users_private_non_workshop_overlay(user_client_with_overlay) -> None:
|
|
with session_scope() as session:
|
|
other = User(username="mallory", password_digest=hash_password("secret"), admin=False)
|
|
session.add(other)
|
|
session.flush()
|
|
overlay = Overlay(
|
|
name="private-external",
|
|
path="private-external",
|
|
type="external",
|
|
user_id=other.id,
|
|
)
|
|
session.add(overlay)
|
|
session.flush()
|
|
overlay_id = overlay.id
|
|
|
|
response = user_client_with_overlay.get(f"/overlays/{overlay_id}")
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_managed_global_overlay_detail_shows_source_url(admin_client) -> None:
|
|
overlay_id = _create_managed_global_overlay()
|
|
|
|
response = admin_client.get(f"/overlays/{overlay_id}")
|
|
text = response.get_data(as_text=True)
|
|
|
|
assert response.status_code == 200
|
|
assert "https://l4d2center.com/maps/servers/index.csv" in text
|
|
|
|
|
|
def test_overlay_update_redirects_to_detail(admin_client) -> None:
|
|
create = admin_client.post(
|
|
"/overlays",
|
|
data={"name": "standard", "type": "external"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert create.status_code == 302
|
|
with session_scope() as session:
|
|
overlay_id = session.query(Overlay).filter_by(name="standard").one().id
|
|
|
|
response = admin_client.post(
|
|
f"/overlays/{overlay_id}",
|
|
data={"name": "renamed"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert response.status_code == 302
|
|
assert response.headers["Location"] == f"/overlays/{overlay_id}"
|
|
|
|
|
|
def test_delete_overlay_rejects_in_use_overlay(admin_client) -> None:
|
|
create = admin_client.post(
|
|
"/overlays",
|
|
data={"name": "standard", "type": "external"},
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
assert create.status_code == 302
|
|
with session_scope() as session:
|
|
overlay_id = session.query(Overlay).filter_by(name="standard").one().id
|
|
|
|
with session_scope() as session:
|
|
user = session.query(User).filter_by(username="admin").one()
|
|
blueprint = Blueprint(user_id=user.id, name="bp", arguments="[]", config="[]")
|
|
session.add(blueprint)
|
|
session.flush()
|
|
session.add(BlueprintOverlay(blueprint_id=blueprint.id, overlay_id=overlay_id, position=0))
|
|
|
|
response = admin_client.post(
|
|
f"/overlays/{overlay_id}/delete",
|
|
headers={"X-CSRF-Token": "test-token"},
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
|
|
|
|
def test_admin_can_enqueue_refresh_global_overlays(admin_client):
|
|
response = admin_client.post("/admin/global-overlays/refresh", headers={"X-CSRF-Token": "test-token"})
|
|
assert response.status_code == 302
|
|
assert response.headers["Location"] == "/admin/jobs"
|