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>
49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
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"
|