left4me/l4d2web/tests/test_global_overlays.py

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="cedapug_maps", 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("workshop", admin=True) is True
    assert is_creatable_overlay_type("external", admin=False) is False
    assert is_creatable_overlay_type("external", admin=True) is False
    assert is_creatable_overlay_type("l4d2center_maps", admin=True) is False
    assert is_creatable_overlay_type("cedapug_maps", admin=True) is False