"""Tests for overlay builders (registry, ExternalBuilder, WorkshopBuilder).""" from __future__ import annotations import os from datetime import UTC, datetime from pathlib import Path import pytest from l4d2web.db import init_db, session_scope from l4d2web.models import Overlay, OverlayWorkshopItem, User, WorkshopItem from l4d2web.services import overlay_builders @pytest.fixture def env(tmp_path, monkeypatch): monkeypatch.setenv("DATABASE_URL", f"sqlite:///{tmp_path/'b.db'}") monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path)) init_db() yield tmp_path def _create_user_and_overlay(name: str, type_: str) -> tuple[int, int]: with session_scope() as s: user = User(username="alice", password_digest="x") s.add(user) s.flush() overlay = Overlay(name=name, path=str(7), type=type_, user_id=user.id) s.add(overlay) s.flush() return user.id, overlay.id def _add_workshop_item(steam_id: str, *, downloaded: bool, cache_root: Path, content: bytes = b"x") -> int: if downloaded: cache_root.mkdir(parents=True, exist_ok=True) (cache_root / f"{steam_id}.vpk").write_bytes(content) with session_scope() as s: wi = WorkshopItem( steam_id=steam_id, title=f"item-{steam_id}", filename=f"orig-{steam_id}.vpk", file_url=f"https://example.com/{steam_id}.vpk", file_size=len(content) if downloaded else 0, time_updated=1700000000 if downloaded else 0, last_downloaded_at=datetime.now(UTC) if downloaded else None, ) s.add(wi) s.flush() return wi.id def _associate(overlay_id: int, item_id: int) -> None: with session_scope() as s: s.add(OverlayWorkshopItem(overlay_id=overlay_id, workshop_item_id=item_id)) def _capture_logs(): out: list[str] = [] err: list[str] = [] return out, err, out.append, err.append def test_registry_has_external_and_workshop() -> None: assert "external" in overlay_builders.BUILDERS assert "workshop" in overlay_builders.BUILDERS def test_registry_unknown_type_raises_keyerror() -> None: with pytest.raises(KeyError): overlay_builders.BUILDERS["nope"] def test_external_builder_is_idempotent_noop_with_log(env: Path) -> None: _, overlay_id = _create_user_and_overlay("ext", "external") out, err, on_stdout, on_stderr = _capture_logs() with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["external"].build( overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False ) assert (env / "overlays" / "7").is_dir() assert any("external overlay" in line for line in out), out # Existing files in the overlay dir are not touched on subsequent build. (env / "overlays" / "7" / "untouched.txt").write_text("data") with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["external"].build( overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False ) assert (env / "overlays" / "7" / "untouched.txt").read_text() == "data" def test_workshop_builder_creates_absolute_symlinks(env: Path) -> None: _, overlay_id = _create_user_and_overlay("ws", "workshop") cache_root = env / "workshop_cache" item_a = _add_workshop_item("1001", downloaded=True, cache_root=cache_root, content=b"AAA") item_b = _add_workshop_item("1002", downloaded=True, cache_root=cache_root, content=b"BBBB") _associate(overlay_id, item_a) _associate(overlay_id, item_b) out, err, on_stdout, on_stderr = _capture_logs() with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["workshop"].build( overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False ) addons = env / "overlays" / "7" / "left4dead2" / "addons" link_a = addons / "1001.vpk" link_b = addons / "1002.vpk" assert link_a.is_symlink() assert link_b.is_symlink() # Targets must be ABSOLUTE so they resolve in the host's namespace. assert os.path.isabs(os.readlink(link_a)) assert os.path.isabs(os.readlink(link_b)) # And they must resolve to the cache files. assert link_a.resolve() == (cache_root / "1001.vpk").resolve() assert link_b.resolve() == (cache_root / "1002.vpk").resolve() def test_workshop_builder_skips_uncached_items_with_warning(env: Path) -> None: _, overlay_id = _create_user_and_overlay("ws", "workshop") cache_root = env / "workshop_cache" cached = _add_workshop_item("1001", downloaded=True, cache_root=cache_root) uncached = _add_workshop_item("9999", downloaded=False, cache_root=cache_root) _associate(overlay_id, cached) _associate(overlay_id, uncached) out, err, on_stdout, on_stderr = _capture_logs() with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["workshop"].build( overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False ) addons = env / "overlays" / "7" / "left4dead2" / "addons" assert (addons / "1001.vpk").is_symlink() assert not (addons / "9999.vpk").exists(), "must NOT create dangling symlink" assert any("9999" in line and ("skip" in line.lower() or "uncached" in line.lower()) for line in err + out), err + out def test_workshop_builder_rerun_is_idempotent(env: Path) -> None: _, overlay_id = _create_user_and_overlay("ws", "workshop") cache_root = env / "workshop_cache" item = _add_workshop_item("1001", downloaded=True, cache_root=cache_root) _associate(overlay_id, item) addons = env / "overlays" / "7" / "left4dead2" / "addons" # First run. with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=lambda: False) first_inode = (addons / "1001.vpk").lstat().st_ino # Second run — no-op. out, err, on_stdout, on_stderr = _capture_logs() with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False) second_inode = (addons / "1001.vpk").lstat().st_ino assert first_inode == second_inode, "symlink should be untouched on idempotent rebuild" assert any("unchanged" in line.lower() for line in out), out def test_workshop_builder_removes_obsolete_symlinks(env: Path) -> None: _, overlay_id = _create_user_and_overlay("ws", "workshop") cache_root = env / "workshop_cache" item_a = _add_workshop_item("1001", downloaded=True, cache_root=cache_root) item_b = _add_workshop_item("1002", downloaded=True, cache_root=cache_root) _associate(overlay_id, item_a) _associate(overlay_id, item_b) addons = env / "overlays" / "7" / "left4dead2" / "addons" with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=lambda: False) assert (addons / "1002.vpk").is_symlink() # Remove the association for 1002. with session_scope() as s: s.query(OverlayWorkshopItem).filter_by(workshop_item_id=item_b).delete() out, err, on_stdout, on_stderr = _capture_logs() with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False) assert (addons / "1001.vpk").is_symlink() assert not (addons / "1002.vpk").exists() # Cache file must remain — overlays are diff-applied, cache is shared. assert (cache_root / "1002.vpk").exists() def test_workshop_builder_leaves_unrelated_files_alone(env: Path) -> None: _, overlay_id = _create_user_and_overlay("ws", "workshop") cache_root = env / "workshop_cache" item = _add_workshop_item("1001", downloaded=True, cache_root=cache_root) _associate(overlay_id, item) addons = env / "overlays" / "7" / "left4dead2" / "addons" addons.mkdir(parents=True, exist_ok=True) (addons / "manual_addon.vpk").write_bytes(b"hand-placed") with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() overlay_builders.BUILDERS["workshop"].build(overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=lambda: False) # Manual file is preserved. assert (addons / "manual_addon.vpk").read_bytes() == b"hand-placed" # Workshop symlink is created alongside. assert (addons / "1001.vpk").is_symlink() def test_workshop_builder_honors_should_cancel(env: Path) -> None: _, overlay_id = _create_user_and_overlay("ws", "workshop") cache_root = env / "workshop_cache" items = [_add_workshop_item(f"100{i}", downloaded=True, cache_root=cache_root) for i in range(3)] for it in items: _associate(overlay_id, it) cancel_calls = {"n": 0} def cancel(): cancel_calls["n"] += 1 return cancel_calls["n"] > 0 # cancel immediately with session_scope() as s: overlay = s.query(Overlay).filter_by(id=overlay_id).one() # Should not crash; partial state is consistent (re-run heals). overlay_builders.BUILDERS["workshop"].build( overlay, on_stdout=lambda _x: None, on_stderr=lambda _x: None, should_cancel=cancel )