left4me/l4d2web/tests/test_overlays.py

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.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 overlay (workshop, no user_id), pre-existing.
        session.add(
            Overlay(name="standard", path="standard", type="workshop", 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_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_workshop_overlay_via_route(admin_client) -> None:
    response = admin_client.post(
        "/overlays",
        data={"name": "standard", "type": "workshop"},
        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_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": "workshop"},
        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 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": "workshop"},
            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": "workshop"},
        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:
    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()

        # Use the seeded system "standard" overlay (id=1).
        overlay_id = session.query(Overlay).filter_by(name="standard").one().id

        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:
    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()
        system_overlay_id = session.query(Overlay).filter_by(name="standard").one().id
        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 "standard" 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_update_redirects_to_detail(admin_client) -> None:
    create = admin_client.post(
        "/overlays",
        data={"name": "standard", "type": "workshop"},
        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": "workshop"},
        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