- {% set can_edit = g.user.admin or (overlay.type == 'workshop' and overlay.user_id == g.user.id) %}
+ {% set can_edit = overlay.type not in ['l4d2center_maps', 'cedapug_maps'] and (g.user.admin or (overlay.type == 'workshop' and overlay.user_id == g.user.id)) %}
{% if can_edit %}
diff --git a/l4d2web/tests/test_blueprints.py b/l4d2web/tests/test_blueprints.py
index 5973a29..3fe17e0 100644
--- a/l4d2web/tests/test_blueprints.py
+++ b/l4d2web/tests/test_blueprints.py
@@ -79,6 +79,86 @@ def test_user_can_create_private_blueprint(user_client) -> None:
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"})
diff --git a/l4d2web/tests/test_global_map_cache.py b/l4d2web/tests/test_global_map_cache.py
new file mode 100644
index 0000000..c5c5d0d
--- /dev/null
+++ b/l4d2web/tests/test_global_map_cache.py
@@ -0,0 +1,49 @@
+from pathlib import Path
+from zipfile import ZipFile
+
+from l4d2web.services.global_map_cache import (
+ extracted_vpk_md5,
+ global_overlay_cache_root,
+ safe_extract_zip_vpks,
+ source_cache_root,
+)
+
+
+def test_global_overlay_cache_paths(tmp_path, monkeypatch):
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+
+ assert global_overlay_cache_root() == tmp_path / "global_overlay_cache"
+ assert source_cache_root("l4d2center-maps") == tmp_path / "global_overlay_cache" / "l4d2center-maps"
+
+
+def test_safe_extract_zip_vpks_extracts_only_vpks(tmp_path):
+ archive = tmp_path / "maps.zip"
+ with ZipFile(archive, "w") as zf:
+ zf.writestr("FatalFreight.vpk", b"vpk-bytes")
+ zf.writestr("readme.txt", b"ignore")
+
+ out_dir = tmp_path / "out"
+ files = safe_extract_zip_vpks(archive, out_dir)
+
+ assert files == [out_dir / "FatalFreight.vpk"]
+ assert (out_dir / "FatalFreight.vpk").read_bytes() == b"vpk-bytes"
+ assert not (out_dir / "readme.txt").exists()
+
+
+def test_safe_extract_zip_vpks_rejects_path_traversal(tmp_path):
+ archive = tmp_path / "bad.zip"
+ with ZipFile(archive, "w") as zf:
+ zf.writestr("../evil.vpk", b"bad")
+
+ try:
+ safe_extract_zip_vpks(archive, tmp_path / "out")
+ except ValueError as exc:
+ assert "unsafe archive member" in str(exc)
+ else:
+ raise AssertionError("path traversal must fail")
+
+
+def test_extracted_vpk_md5(tmp_path):
+ p = tmp_path / "x.vpk"
+ p.write_bytes(b"abc")
+ assert extracted_vpk_md5(p) == "900150983cd24fb0d6963f7d28e17f72"
diff --git a/l4d2web/tests/test_global_map_sources.py b/l4d2web/tests/test_global_map_sources.py
new file mode 100644
index 0000000..99cf846
--- /dev/null
+++ b/l4d2web/tests/test_global_map_sources.py
@@ -0,0 +1,65 @@
+from l4d2web.services.global_map_sources import (
+ GlobalMapManifestItem,
+ parse_cedapug_custom_html,
+ parse_l4d2center_csv,
+)
+
+
+def test_parse_l4d2center_csv_semicolon_manifest():
+ raw = """Name;Size;md5;Download link
+carriedoff.vpk;128660532;0380e12c57156574e17a96da1252cf21;https://l4d2center.com/maps/servers/carriedoff.7z
+"""
+
+ items = parse_l4d2center_csv(raw)
+
+ assert items == [
+ GlobalMapManifestItem(
+ item_key="carriedoff.vpk",
+ display_name="carriedoff.vpk",
+ download_url="https://l4d2center.com/maps/servers/carriedoff.7z",
+ expected_vpk_name="carriedoff.vpk",
+ expected_size=128660532,
+ expected_md5="0380e12c57156574e17a96da1252cf21",
+ )
+ ]
+
+
+def test_parse_l4d2center_rejects_missing_header():
+ try:
+ parse_l4d2center_csv("bad,data\n")
+ except ValueError as exc:
+ assert "Name;Size;md5;Download link" in str(exc)
+ else:
+ raise AssertionError("bad header must fail")
+
+
+def test_parse_cedapug_custom_html_extracts_relative_zip_links():
+ html = """
+
+ """
+
+ items = parse_cedapug_custom_html(html)
+
+ assert items == [
+ GlobalMapManifestItem(
+ item_key="FatalFreight.zip",
+ display_name="Fatal Freight",
+ download_url="https://cedapug.com/maps/FatalFreight.zip",
+ expected_vpk_name="",
+ expected_size=None,
+ expected_md5="",
+ )
+ ]
+
+
+def test_parse_cedapug_custom_html_rejects_missing_data():
+ try:
+ parse_cedapug_custom_html("")
+ except ValueError as exc:
+ assert "renderCustomMapDownloads" in str(exc)
+ else:
+ raise AssertionError("missing embedded data must fail")
diff --git a/l4d2web/tests/test_global_overlay_builders.py b/l4d2web/tests/test_global_overlay_builders.py
new file mode 100644
index 0000000..0cde425
--- /dev/null
+++ b/l4d2web/tests/test_global_overlay_builders.py
@@ -0,0 +1,89 @@
+import os
+from pathlib import Path
+
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import GlobalOverlayItem, GlobalOverlayItemFile, GlobalOverlaySource, Overlay
+from l4d2web.services.overlay_builders import BUILDERS
+
+
+def seed_source(tmp_path: Path, monkeypatch) -> int:
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'builder.db'}")
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+ init_db()
+ cache_vpk = tmp_path / "global_overlay_cache" / "l4d2center-maps" / "vpks" / "carriedoff.vpk"
+ cache_vpk.parent.mkdir(parents=True, exist_ok=True)
+ cache_vpk.write_bytes(b"vpk")
+ with session_scope() as db:
+ overlay = Overlay(name="l4d2center-maps", path="7", type="l4d2center_maps", user_id=None)
+ db.add(overlay)
+ db.flush()
+ source = GlobalOverlaySource(
+ overlay_id=overlay.id,
+ source_key="l4d2center-maps",
+ source_type="l4d2center_csv",
+ source_url="https://l4d2center.com/maps/servers/index.csv",
+ )
+ db.add(source)
+ db.flush()
+ item = GlobalOverlayItem(
+ source_id=source.id,
+ item_key="carriedoff.vpk",
+ display_name="carriedoff.vpk",
+ download_url="https://example.invalid/carriedoff.7z",
+ expected_vpk_name="carriedoff.vpk",
+ )
+ db.add(item)
+ db.flush()
+ db.add(
+ GlobalOverlayItemFile(
+ item_id=item.id,
+ vpk_name="carriedoff.vpk",
+ cache_path="l4d2center-maps/vpks/carriedoff.vpk",
+ size=3,
+ md5="",
+ )
+ )
+ db.flush()
+ return overlay.id
+
+
+def test_registry_contains_global_map_builders():
+ assert "l4d2center_maps" in BUILDERS
+ assert "cedapug_maps" in BUILDERS
+
+
+def test_global_builder_creates_absolute_symlink(tmp_path, monkeypatch):
+ overlay_id = seed_source(tmp_path, monkeypatch)
+ out: list[str] = []
+ err: list[str] = []
+ with session_scope() as db:
+ overlay = db.query(Overlay).filter_by(id=overlay_id).one()
+ BUILDERS["l4d2center_maps"].build(overlay, on_stdout=out.append, on_stderr=err.append, should_cancel=lambda: False)
+
+ link = tmp_path / "overlays" / "7" / "left4dead2" / "addons" / "carriedoff.vpk"
+ assert link.is_symlink()
+ assert os.path.isabs(os.readlink(link))
+ assert link.resolve() == (tmp_path / "global_overlay_cache" / "l4d2center-maps" / "vpks" / "carriedoff.vpk").resolve()
+ assert any("global overlay" in line for line in out)
+
+
+def test_global_builder_removes_obsolete_managed_symlink_but_keeps_foreign(tmp_path, monkeypatch):
+ overlay_id = seed_source(tmp_path, monkeypatch)
+ addons = tmp_path / "overlays" / "7" / "left4dead2" / "addons"
+ addons.mkdir(parents=True, exist_ok=True)
+ foreign_target = tmp_path / "foreign.vpk"
+ foreign_target.write_bytes(b"foreign")
+ os.symlink(str(foreign_target), addons / "foreign.vpk")
+
+ with session_scope() as db:
+ overlay = db.query(Overlay).filter_by(id=overlay_id).one()
+ BUILDERS["l4d2center_maps"].build(overlay, on_stdout=lambda line: None, on_stderr=lambda line: None, should_cancel=lambda: False)
+ source = db.query(GlobalOverlaySource).filter_by(source_key="l4d2center-maps").one()
+ db.query(GlobalOverlayItem).filter_by(source_id=source.id).delete()
+
+ with session_scope() as db:
+ overlay = db.query(Overlay).filter_by(id=overlay_id).one()
+ BUILDERS["l4d2center_maps"].build(overlay, on_stdout=lambda line: None, on_stderr=lambda line: None, should_cancel=lambda: False)
+
+ assert not (addons / "carriedoff.vpk").exists()
+ assert (addons / "foreign.vpk").is_symlink()
diff --git a/l4d2web/tests/test_global_overlay_cli.py b/l4d2web/tests/test_global_overlay_cli.py
new file mode 100644
index 0000000..13dc429
--- /dev/null
+++ b/l4d2web/tests/test_global_overlay_cli.py
@@ -0,0 +1,19 @@
+from l4d2web.app import create_app
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import Job
+
+
+def test_refresh_global_overlays_cli_enqueues_system_job(tmp_path, monkeypatch):
+ db_url = f"sqlite:///{tmp_path/'cli.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()
+
+ result = app.test_cli_runner().invoke(args=["refresh-global-overlays"])
+
+ assert result.exit_code == 0
+ assert "queued refresh_global_overlays job" in result.output
+ with session_scope() as db:
+ job = db.query(Job).filter_by(operation="refresh_global_overlays").one()
+ assert job.user_id is None
diff --git a/l4d2web/tests/test_global_overlay_models.py b/l4d2web/tests/test_global_overlay_models.py
new file mode 100644
index 0000000..3c94f7b
--- /dev/null
+++ b/l4d2web/tests/test_global_overlay_models.py
@@ -0,0 +1,154 @@
+from sqlalchemy.exc import IntegrityError
+
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import (
+ GlobalOverlayItem,
+ GlobalOverlayItemFile,
+ GlobalOverlaySource,
+ Job,
+ Overlay,
+ User,
+)
+
+
+def test_system_job_allows_null_user_id(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'models.db'}")
+ init_db()
+
+ with session_scope() as db:
+ job = Job(
+ user_id=None,
+ server_id=None,
+ overlay_id=None,
+ operation="refresh_global_overlays",
+ )
+ db.add(job)
+ db.flush()
+ assert job.id is not None
+ assert job.user_id is None
+
+
+def test_global_overlay_source_uniqueness(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'sources.db'}")
+ init_db()
+
+ with session_scope() as db:
+ overlay = Overlay(
+ name="l4d2center-maps", path="1", type="l4d2center_maps", user_id=None
+ )
+ db.add(overlay)
+ db.flush()
+ db.add(
+ GlobalOverlaySource(
+ overlay_id=overlay.id,
+ source_key="l4d2center-maps",
+ source_type="l4d2center_csv",
+ source_url="https://l4d2center.com/maps/servers/index.csv",
+ )
+ )
+
+ try:
+ with session_scope() as db:
+ other = Overlay(
+ name="cedapug-maps", path="2", type="cedapug_maps", user_id=None
+ )
+ db.add(other)
+ db.flush()
+ db.add(
+ GlobalOverlaySource(
+ overlay_id=other.id,
+ source_key="l4d2center-maps",
+ source_type="l4d2center_csv",
+ source_url="https://example.invalid/duplicate",
+ )
+ )
+ except IntegrityError:
+ pass
+ else:
+ raise AssertionError("duplicate source_key must fail")
+
+
+def test_global_overlay_items_and_files_are_unique_per_parent(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'items.db'}")
+ init_db()
+
+ with session_scope() as db:
+ overlay = Overlay(name="cedapug-maps", path="1", type="cedapug_maps", user_id=None)
+ db.add(overlay)
+ db.flush()
+ source = GlobalOverlaySource(
+ overlay_id=overlay.id,
+ source_key="cedapug-maps",
+ source_type="cedapug_custom_page",
+ source_url="https://cedapug.com/custom",
+ )
+ db.add(source)
+ db.flush()
+ item = GlobalOverlayItem(
+ source_id=source.id,
+ item_key="FatalFreight.zip",
+ display_name="Fatal Freight",
+ download_url="https://cedapug.com/maps/FatalFreight.zip",
+ expected_vpk_name="FatalFreight.vpk",
+ )
+ db.add(item)
+ db.flush()
+ db.add(
+ GlobalOverlayItemFile(
+ item_id=item.id,
+ vpk_name="FatalFreight.vpk",
+ cache_path="cedapug-maps/vpks/FatalFreight.vpk",
+ size=123,
+ md5="",
+ )
+ )
+ item_id = item.id
+
+ try:
+ with session_scope() as db:
+ source = db.query(GlobalOverlaySource).filter_by(source_key="cedapug-maps").one()
+ db.add(
+ GlobalOverlayItem(
+ source_id=source.id,
+ item_key="FatalFreight.zip",
+ display_name="Fatal Freight duplicate",
+ download_url="https://cedapug.com/maps/FatalFreight.zip",
+ expected_vpk_name="FatalFreight.vpk",
+ )
+ )
+ except IntegrityError:
+ pass
+ else:
+ raise AssertionError("duplicate item_key per source must fail")
+
+ try:
+ with session_scope() as db:
+ db.add(
+ GlobalOverlayItemFile(
+ item_id=item_id,
+ vpk_name="FatalFreight.vpk",
+ cache_path="cedapug-maps/vpks/FatalFreight-copy.vpk",
+ size=456,
+ md5="",
+ )
+ )
+ except IntegrityError:
+ pass
+ else:
+ raise AssertionError("duplicate vpk_name per item must fail")
+
+
+def test_normal_user_rows_still_require_real_users(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'users.db'}")
+ init_db()
+
+ with session_scope() as db:
+ user = User(username="alice", password_digest="digest", admin=False)
+ db.add(user)
+ db.flush()
+ job = Job(user_id=user.id, server_id=None, operation="install", state="queued")
+ db.add(job)
+ db.flush()
+
+ assert job.id is not None
+ assert job.user_id == user.id
diff --git a/l4d2web/tests/test_global_overlay_refresh.py b/l4d2web/tests/test_global_overlay_refresh.py
new file mode 100644
index 0000000..bed1fe2
--- /dev/null
+++ b/l4d2web/tests/test_global_overlay_refresh.py
@@ -0,0 +1,69 @@
+from pathlib import Path
+
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import GlobalOverlayItem, GlobalOverlayItemFile, GlobalOverlaySource
+from l4d2web.services.global_map_sources import GlobalMapManifestItem
+
+
+def test_refresh_global_overlays_updates_manifest_items_and_invokes_builders(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'refresh.db'}")
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+ init_db()
+
+ from l4d2web.services import global_overlay_refresh
+ monkeypatch.setattr(
+ global_overlay_refresh,
+ "fetch_l4d2center_manifest",
+ lambda: ("hash-center", [GlobalMapManifestItem("carriedoff.vpk", "carriedoff.vpk", "https://example.invalid/carriedoff.7z", "carriedoff.vpk", 3, "" )]),
+ )
+ monkeypatch.setattr(
+ global_overlay_refresh,
+ "fetch_cedapug_manifest",
+ lambda: ("hash-ceda", [GlobalMapManifestItem("FatalFreight.zip", "Fatal Freight", "https://example.invalid/FatalFreight.zip")]),
+ )
+
+ def fake_download_and_extract(source_key, item, *, should_cancel):
+ target = tmp_path / "global_overlay_cache" / source_key / "vpks" / (item.expected_vpk_name or item.item_key.replace(".zip", ".vpk"))
+ target.parent.mkdir(parents=True, exist_ok=True)
+ target.write_bytes(b"vpk")
+ return [(target.name, f"{source_key}/vpks/{target.name}", 3, "")], "etag", "last-modified", 3
+
+ built: list[str] = []
+ monkeypatch.setattr(global_overlay_refresh, "download_and_extract_item", fake_download_and_extract)
+ monkeypatch.setattr(global_overlay_refresh, "build_global_overlay", lambda overlay, **kwargs: built.append(overlay.name))
+
+ out: list[str] = []
+ result = global_overlay_refresh.refresh_global_overlays(on_stdout=out.append, on_stderr=out.append, should_cancel=lambda: False)
+
+ assert result == ["cedapug-maps", "l4d2center-maps"]
+ assert set(built) == {"cedapug-maps", "l4d2center-maps"}
+ with session_scope() as db:
+ assert db.query(GlobalOverlaySource).count() == 2
+ assert db.query(GlobalOverlayItem).count() == 2
+ assert db.query(GlobalOverlayItemFile).count() == 2
+
+
+def test_refresh_removes_items_absent_from_manifest(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'remove.db'}")
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+ init_db()
+
+ from l4d2web.services.global_overlays import ensure_global_overlays
+ from l4d2web.services import global_overlay_refresh
+
+ with session_scope() as db:
+ ensure_global_overlays(db)
+ source = db.query(GlobalOverlaySource).filter_by(source_key="l4d2center-maps").one()
+ item = GlobalOverlayItem(source_id=source.id, item_key="old.vpk", display_name="old.vpk", download_url="https://example.invalid/old.7z")
+ db.add(item)
+ db.flush()
+ db.add(GlobalOverlayItemFile(item_id=item.id, vpk_name="old.vpk", cache_path="l4d2center-maps/vpks/old.vpk", size=3))
+
+ monkeypatch.setattr(global_overlay_refresh, "fetch_l4d2center_manifest", lambda: ("empty-center", []))
+ monkeypatch.setattr(global_overlay_refresh, "fetch_cedapug_manifest", lambda: ("empty-ceda", []))
+ monkeypatch.setattr(global_overlay_refresh, "build_global_overlay", lambda overlay, **kwargs: None)
+
+ global_overlay_refresh.refresh_global_overlays(on_stdout=lambda line: None, on_stderr=lambda line: None, should_cancel=lambda: False)
+
+ with session_scope() as db:
+ assert db.query(GlobalOverlayItem).filter_by(item_key="old.vpk").count() == 0
diff --git a/l4d2web/tests/test_global_overlays.py b/l4d2web/tests/test_global_overlays.py
new file mode 100644
index 0000000..011c89b
--- /dev/null
+++ b/l4d2web/tests/test_global_overlays.py
@@ -0,0 +1,167 @@
+from sqlalchemy import select
+
+from l4d2web.db import init_db, session_scope
+from l4d2web.models import GlobalOverlaySource, Job, Overlay, User
+from l4d2web.services.global_overlays import (
+ enqueue_refresh_global_overlays,
+ ensure_global_overlays,
+ is_creatable_overlay_type,
+)
+
+
+def test_ensure_global_overlays_creates_singletons_and_directories(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'global_overlays.db'}")
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+ init_db()
+
+ with session_scope() as session:
+ created = ensure_global_overlays(session)
+ assert created == {"cedapug-maps", "l4d2center-maps"}
+
+ second = ensure_global_overlays(session)
+ assert second == set()
+
+ overlays = session.scalars(select(Overlay).order_by(Overlay.name)).all()
+ assert [overlay.name for overlay in overlays] == ["cedapug-maps", "l4d2center-maps"]
+ assert [overlay.type for overlay in overlays] == ["cedapug_maps", "l4d2center_maps"]
+ assert [overlay.user_id for overlay in overlays] == [None, None]
+ assert len({overlay.path for overlay in overlays}) == 2
+ for overlay in overlays:
+ assert (tmp_path / "overlays" / overlay.path).is_dir()
+
+ sources = session.scalars(select(GlobalOverlaySource).order_by(GlobalOverlaySource.source_key)).all()
+ assert [source.source_key for source in sources] == ["cedapug-maps", "l4d2center-maps"]
+ assert [source.source_type for source in sources] == [
+ "cedapug_custom_page",
+ "l4d2center_csv",
+ ]
+
+
+def test_ensure_global_overlays_repairs_existing_rows(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'global_overlay_repair.db'}")
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+ init_db()
+
+ with session_scope() as session:
+ overlay = Overlay(name="cedapug-maps", path="legacy", type="external", user_id=None)
+ session.add(overlay)
+ session.flush()
+ session.add(
+ GlobalOverlaySource(
+ overlay_id=overlay.id,
+ source_key="cedapug-maps",
+ source_type="wrong",
+ source_url="https://example.invalid/wrong",
+ )
+ )
+
+ (tmp_path / "overlays" / "legacy").mkdir(parents=True)
+
+ with session_scope() as session:
+ created = ensure_global_overlays(session)
+ assert created == {"l4d2center-maps"}
+
+ repaired = session.scalar(select(Overlay).where(Overlay.name == "cedapug-maps"))
+ assert repaired is not None
+ assert repaired.type == "cedapug_maps"
+ assert repaired.user_id is None
+ assert (tmp_path / "overlays" / repaired.path).is_dir()
+
+ source = session.scalar(
+ select(GlobalOverlaySource).where(GlobalOverlaySource.source_key == "cedapug-maps")
+ )
+ assert source is not None
+ assert source.source_type == "cedapug_custom_page"
+ assert source.source_url == "https://cedapug.com/custom"
+
+
+def test_ensure_global_overlays_does_not_hijack_private_overlay_name(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'global_overlay_private_name.db'}")
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+ init_db()
+
+ with session_scope() as session:
+ user = User(username="alice", password_digest="digest", admin=False)
+ session.add(user)
+ session.flush()
+ private = Overlay(
+ name="l4d2center-maps",
+ path="private-l4d2center",
+ type="workshop",
+ user_id=user.id,
+ )
+ session.add(private)
+ session.flush()
+ private_id = private.id
+ private_user_id = user.id
+
+ with session_scope() as session:
+ created = ensure_global_overlays(session)
+
+ assert created == {"cedapug-maps", "l4d2center-maps"}
+ private = session.scalar(select(Overlay).where(Overlay.id == private_id))
+ assert private is not None
+ assert private.user_id == private_user_id
+ assert private.type == "workshop"
+ assert private.path == "private-l4d2center"
+
+ system = session.scalar(
+ select(Overlay).where(Overlay.name == "l4d2center-maps", Overlay.user_id.is_(None))
+ )
+ assert system is not None
+ assert system.id != private_id
+ assert system.type == "l4d2center_maps"
+ assert (tmp_path / "overlays" / system.path).is_dir()
+
+ source = session.scalar(
+ select(GlobalOverlaySource).where(GlobalOverlaySource.source_key == "l4d2center-maps")
+ )
+ assert source is not None
+ assert source.overlay_id == system.id
+
+
+def test_enqueue_refresh_global_overlays_coalesces_active_jobs(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'refresh_jobs.db'}")
+ init_db()
+
+ for state in ("queued", "running", "cancelling"):
+ with session_scope() as session:
+ session.query(Job).delete()
+ existing = Job(
+ user_id=7,
+ server_id=None,
+ overlay_id=None,
+ operation="refresh_global_overlays",
+ state=state,
+ )
+ session.add(existing)
+ session.flush()
+ existing_id = existing.id
+
+ job = enqueue_refresh_global_overlays(session, user_id=None)
+ assert job.id == existing_id
+ assert session.query(Job).filter_by(operation="refresh_global_overlays").count() == 1
+
+
+def test_enqueue_refresh_global_overlays_creates_system_job(tmp_path, monkeypatch):
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'refresh_system_job.db'}")
+ init_db()
+
+ with session_scope() as session:
+ job = enqueue_refresh_global_overlays(session, user_id=None)
+
+ assert job.id is not None
+ assert job.user_id is None
+ assert job.server_id is None
+ assert job.overlay_id is None
+ assert job.operation == "refresh_global_overlays"
+ assert job.state == "queued"
+
+
+def test_is_creatable_overlay_type_policy():
+ assert is_creatable_overlay_type("workshop", admin=False) is True
+ assert is_creatable_overlay_type("external", admin=False) is False
+ assert is_creatable_overlay_type("external", admin=True) is True
+ assert is_creatable_overlay_type("workshop", admin=True) is True
+ assert is_creatable_overlay_type("l4d2center_maps", admin=True) is False
+ assert is_creatable_overlay_type("cedapug_maps", admin=True) is False
diff --git a/l4d2web/tests/test_job_logs.py b/l4d2web/tests/test_job_logs.py
index 0982b19..0168a8a 100644
--- a/l4d2web/tests/test_job_logs.py
+++ b/l4d2web/tests/test_job_logs.py
@@ -104,3 +104,28 @@ def test_sse_js_handles_job_log_custom_events() -> None:
assert 'addEventListener("stdout"' in js
assert 'addEventListener("stderr"' in js
+
+
+def test_system_job_logs_persist(tmp_path, monkeypatch):
+ from l4d2web.models import Job, JobLog
+ from l4d2web.services.job_worker import append_job_log
+
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'system-job-logs.db'}")
+ init_db()
+
+ with session_scope() as db:
+ job = Job(
+ user_id=None,
+ server_id=None,
+ operation="refresh_global_overlays",
+ state="queued",
+ )
+ db.add(job)
+ db.flush()
+
+ seq = append_job_log(db, job.id, "stdout", "queued by system timer")
+ db.flush()
+
+ row = db.query(JobLog).filter_by(job_id=job.id).one()
+ assert seq == 1
+ assert row.line == "queued by system timer"
diff --git a/l4d2web/tests/test_job_worker.py b/l4d2web/tests/test_job_worker.py
index d6bbc0e..0d0cd7f 100644
--- a/l4d2web/tests/test_job_worker.py
+++ b/l4d2web/tests/test_job_worker.py
@@ -700,3 +700,48 @@ def test_refresh_job_enqueues_build_overlay_without_locking_its_final_log(
assert job.state == "succeeded"
assert build_job is not None
assert "enqueued build_overlay for 1 overlay(s)" in lines
+
+
+def test_refresh_global_overlays_blocks_install_build_refresh_and_servers() -> None:
+ from l4d2web.services.job_worker import SchedulerState, can_start
+
+ state = SchedulerState(refresh_global_overlays_running=True)
+ assert can_start(DummyJob(operation="install"), state) is False
+ assert can_start(DummyJob(operation="refresh_workshop_items"), state) is False
+ assert can_start(DummyJob(operation="build_overlay", overlay_id=1), state) is False
+ assert can_start(DummyJob(operation="start", server_id=1), state) is False
+
+
+def test_refresh_global_overlays_waits_for_active_work() -> None:
+ from l4d2web.services.job_worker import SchedulerState, can_start
+
+ assert can_start(DummyJob(operation="refresh_global_overlays"), SchedulerState(install_running=True)) is False
+ assert can_start(DummyJob(operation="refresh_global_overlays"), SchedulerState(refresh_running=True)) is False
+ state = SchedulerState()
+ state.running_overlays.add(1)
+ assert can_start(DummyJob(operation="refresh_global_overlays"), state) is False
+ state = SchedulerState()
+ state.running_servers.add(1)
+ assert can_start(DummyJob(operation="refresh_global_overlays"), state) is False
+
+
+def test_run_worker_once_dispatches_refresh_global_overlays(seeded_worker, monkeypatch):
+ from l4d2web.services import job_worker
+ from l4d2web.models import Job
+ from l4d2web.db import session_scope
+
+ called = []
+
+ def fake_refresh(*, on_stdout, on_stderr, should_cancel):
+ called.append("refresh")
+ on_stdout("global refresh complete")
+ return ["l4d2center-maps"]
+
+ monkeypatch.setattr(job_worker, "_run_refresh_global_overlays", fake_refresh)
+ with session_scope() as db:
+ job = Job(user_id=None, server_id=None, operation="refresh_global_overlays", state="queued")
+ db.add(job)
+
+ app, ids = seeded_worker
+ assert job_worker.run_worker_once() is True
+ assert called == ["refresh"]
diff --git a/l4d2web/tests/test_l4d2_facade.py b/l4d2web/tests/test_l4d2_facade.py
index 30fed82..aa9e903 100644
--- a/l4d2web/tests/test_l4d2_facade.py
+++ b/l4d2web/tests/test_l4d2_facade.py
@@ -258,3 +258,55 @@ def test_initialize_fails_fast_on_uncached_workshop_items(
assert str(overlay_id) in msg or "ws" in msg
# l4d2ctl initialize MUST NOT run when uncached items are present.
assert all("initialize" not in cmd for cmd in invocations), invocations
+
+
+def test_initialize_fails_when_global_overlay_cache_file_missing(tmp_path, monkeypatch):
+ from l4d2web.db import init_db, session_scope
+ from l4d2web.models import (
+ Blueprint,
+ BlueprintOverlay,
+ GlobalOverlayItem,
+ GlobalOverlayItemFile,
+ GlobalOverlaySource,
+ Overlay,
+ Server,
+ User,
+ )
+ from l4d2web.services.l4d2_facade import initialize_server
+
+ monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'facade-global.db'}")
+ monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
+ init_db()
+
+ with session_scope() as db:
+ user = User(username="alice", password_digest="digest")
+ db.add(user)
+ db.flush()
+ overlay = Overlay(name="l4d2center-maps", path="7", type="l4d2center_maps", user_id=None)
+ db.add(overlay)
+ db.flush()
+ source = GlobalOverlaySource(overlay_id=overlay.id, source_key="l4d2center-maps", source_type="l4d2center_csv", source_url="https://l4d2center.com/maps/servers/index.csv")
+ db.add(source)
+ db.flush()
+ item = GlobalOverlayItem(source_id=source.id, item_key="carriedoff.vpk", display_name="carriedoff.vpk", download_url="https://example.invalid/carriedoff.7z")
+ db.add(item)
+ db.flush()
+ db.add(GlobalOverlayItemFile(item_id=item.id, vpk_name="carriedoff.vpk", cache_path="l4d2center-maps/vpks/carriedoff.vpk", size=123))
+ blueprint = Blueprint(user_id=user.id, name="bp", arguments="[]", config="[]")
+ db.add(blueprint)
+ db.flush()
+ db.add(BlueprintOverlay(blueprint_id=blueprint.id, overlay_id=overlay.id, position=0))
+ server = Server(user_id=user.id, blueprint_id=blueprint.id, name="alpha", port=27015)
+ db.add(server)
+ db.flush()
+ server_id = server.id
+
+ monkeypatch.setattr("l4d2web.services.host_commands.run_command", lambda *args, **kwargs: None)
+
+ try:
+ initialize_server(server_id)
+ except RuntimeError as exc:
+ assert "carriedoff.vpk" in str(exc)
+ assert "l4d2center-maps" in str(exc)
+ else:
+ raise AssertionError("missing global overlay cache file must fail")
diff --git a/l4d2web/tests/test_overlays.py b/l4d2web/tests/test_overlays.py
index 1825ba1..e4f2c4f 100644
--- a/l4d2web/tests/test_overlays.py
+++ b/l4d2web/tests/test_overlays.py
@@ -2,7 +2,7 @@ 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, User
+from l4d2web.models import Blueprint, BlueprintOverlay, GlobalOverlaySource, Overlay, User
from l4d2web.services.security import validate_overlay_ref
@@ -60,6 +60,16 @@ def test_user_can_view_overlay_catalog(user_client_with_overlay) -> None:
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)
@@ -80,6 +90,16 @@ def test_admin_can_create_external_overlay(admin_client) -> None:
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):
@@ -92,7 +112,7 @@ def test_non_admin_cannot_create_external_overlay(user_client_with_overlay) -> N
data={"name": "bad", "type": "external"},
headers={"X-CSRF-Token": "test-token"},
)
- assert response.status_code == 403
+ assert response.status_code == 400
def test_user_can_create_workshop_overlay(user_client_with_overlay) -> None:
@@ -184,6 +204,62 @@ def test_admin_can_update_and_delete_overlay(admin_client) -> None:
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"):
@@ -235,6 +311,56 @@ def test_overlay_detail_page_lists_using_blueprints(admin_client) -> None:
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
@@ -251,6 +377,36 @@ def test_overlay_detail_hides_edit_for_non_admin_external(user_client_with_overl
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",
@@ -293,3 +449,9 @@ def test_delete_overlay_rejects_in_use_overlay(admin_client) -> None:
)
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"
diff --git a/l4d2web/tests/test_pages.py b/l4d2web/tests/test_pages.py
index 82f7a3e..a5c7047 100644
--- a/l4d2web/tests/test_pages.py
+++ b/l4d2web/tests/test_pages.py
@@ -457,3 +457,56 @@ def test_blueprint_detail_has_ordered_overlay_form(auth_client_with_server) -> N
assert 'name="config"' in text
assert 'name="overlay_ids"' in text
assert 'name="overlay_position_1"' in text
+
+
+def test_admin_jobs_page_renders_system_job(tmp_path, monkeypatch) -> None:
+ db_url = f"sqlite:///{tmp_path/'admin-system-job.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:
+ admin = User(username="admin", password_digest=hash_password("secret"), admin=True)
+ session.add(admin)
+ session.flush()
+ admin_id = admin.id
+
+ admin_client = app.test_client()
+ with admin_client.session_transaction() as sess:
+ sess["user_id"] = admin_id
+
+ with session_scope() as db:
+ db.add(Job(user_id=None, server_id=None, operation="refresh_global_overlays", state="queued"))
+
+ response = admin_client.get("/admin/jobs")
+ text = response.get_data(as_text=True)
+
+ assert response.status_code == 200
+ assert "refresh_global_overlays" in text
+ assert "system" in text
+
+
+def test_non_admin_cannot_view_system_job(tmp_path, monkeypatch) -> None:
+ db_url = f"sqlite:///{tmp_path/'non-admin-system-job.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
+
+ user_client = app.test_client()
+ with user_client.session_transaction() as sess:
+ sess["user_id"] = user_id
+
+ with session_scope() as db:
+ job = Job(user_id=None, server_id=None, operation="refresh_global_overlays", state="queued")
+ db.add(job)
+ db.flush()
+ job_id = job.id
+
+ response = user_client.get(f"/jobs/{job_id}")
+ assert response.status_code == 403