Forge-Engine/tests/game/test_custom_events.py

"""Événements personnalisés (db/custom_events/, game_engine/custom_events/) :
un événement game-wide (ex. NEED_ACTION) écouté par un nœud Déclencheur
("trigger_event"="evenement") sur UNE scène/UN modèle, et déclenché par
un nœud Action ("action_type"="declencher_evenement") sur UNE AUTRE
scène/UN AUTRE modèle — la seule entité game-wide du moteur dont la
suppression cascade RÉELLEMENT vers les nœuds qui la référencent,
partout à la fois (voir game_engine/custom_events/delete_custom_event.py).

Un événement est une pure NOTIFICATION, sans aucun paramètre : le
déclencher ne fait que signaler, jamais choisir un élément — c'est à
l'ÉCOUTEUR (déclencheur → condition → action) de décider quoi faire
ensuite, avec ses propres réglages habituels (cible fixe, "Ligne
cliquée"...)."""

import json
import re
from typing import Any

from flask.testing import FlaskClient

import db
import game_engine
from tests.conftest import not_none


def _create_screen(client: FlaskClient, slug: str, name: str = "Accueil") -> Any:
    resp = client.post(f"/game/{slug}/screens/new", data={"name": name}, follow_redirects=False)
    return int(not_none(re.search(r"/screens/(\d+)/edit", resp.headers["Location"])).group(1))


def _add_flow_node(client: FlaskClient, slug: str, screen_id: int, **fields: Any) -> Any:
    resp = client.post(
        f"/game/{slug}/screens/{screen_id}/flow/nodes/add",
        data=json.dumps(fields),
        content_type="application/json",
    )
    assert resp.status_code == 200
    return resp.get_json()["id"]


def test_create_event_is_idempotent_by_name(client: FlaskClient, game: str) -> None:
    screen_id = _create_screen(client, game)
    resp = client.post(
        f"/game/{game}/events/new",
        data={
            "name": "NEED_ACTION",
            "description": "Besoin d'action",
            "screen_id": screen_id,
        },
    )
    assert resp.status_code == 302
    events = db.list_custom_events(game)
    assert len(events) == 1
    assert events[0]["name"] == "NEED_ACTION"

    # Double soumission : ne crée pas un deuxième événement du même nom.
    client.post(f"/game/{game}/events/new", data={"name": "NEED_ACTION", "screen_id": screen_id})
    assert len(db.list_custom_events(game)) == 1


def test_edit_event_updates_description(client: FlaskClient, game: str) -> None:
    screen_id = _create_screen(client, game)
    event_id = db.create_custom_event(game, "NEED_ACTION", "old")
    resp = client.post(
        f"/game/{game}/events/{event_id}/edit",
        data={
            "description": "new",
            "screen_id": screen_id,
        },
    )
    assert resp.status_code == 302
    event = not_none(db.get_custom_event(game, event_id))
    assert event["description"] == "new"


def test_usages_reports_listener_and_dispatcher_across_different_screens(client: FlaskClient, game: str) -> None:
    event_id = not_none(db.create_custom_event(game, "NEED_ACTION"))

    listener_screen = _create_screen(client, game, "Écran écouteur")
    _add_flow_node(
        client, game, listener_screen, node_type="trigger", trigger_event="evenement", trigger_custom_event_id=event_id
    )

    dispatcher_screen = _create_screen(client, game, "Écran déclencheur")
    _add_flow_node(
        client,
        game,
        dispatcher_screen,
        node_type="action",
        action_type="declencher_evenement",
        target_custom_event_id=event_id,
    )

    usages = game_engine.list_custom_event_usages(game)
    roles = {u["role"] for u in usages[event_id]}
    screen_ids = {u["screen_id"] for u in usages[event_id]}
    assert roles == {"ecoute", "declenche"}
    assert screen_ids == {listener_screen, dispatcher_screen}


def test_deleting_an_event_cascades_to_every_referencing_node_everywhere(client: FlaskClient, game: str) -> None:
    event_id = db.create_custom_event(game, "NEED_ACTION")

    listener_screen = _create_screen(client, game, "Écran écouteur")
    listener_node_id = _add_flow_node(
        client,
        game,
        listener_screen,
        node_type="trigger",
        trigger_event="evenement",
        trigger_custom_event_id=event_id,
    )

    # Une AUTRE scène — la cascade doit atteindre TOUS les écrans, pas
    # seulement celui qui a servi à déclencher la suppression.
    dispatcher_screen = _create_screen(client, game, "Écran déclencheur")
    dispatcher_node_id = _add_flow_node(
        client,
        game,
        dispatcher_screen,
        node_type="action",
        action_type="declencher_evenement",
        target_custom_event_id=event_id,
    )

    resp = client.post(
        f"/game/{game}/events/{event_id}/delete", data={"screen_id": listener_screen}, follow_redirects=False
    )
    assert resp.status_code == 302

    assert db.get_custom_event(game, event_id) is None
    assert game_engine.get_flow_node(game, listener_node_id) is None
    assert game_engine.get_flow_node(game, dispatcher_node_id) is None
    # Les écrans eux-mêmes ne sont pas touchés, seuls les nœuds.
    assert game_engine.get_screen(game, listener_screen) is not None
    assert game_engine.get_screen(game, dispatcher_screen) is not None


def test_runtime_payload_exposes_both_nodes_for_the_client_js(client: FlaskClient, game: str) -> None:
    """templates/game/play.html::dispatchGameEvent()/findTriggerNode() scannent
    gameData.flows (toutes les scènes ET tous les modèles à la fois) pour
    retrouver un nœud par trigger_custom_event_id/target_custom_event_id —
    ce test verrouille que full_game_payload() expose bien ces deux champs
    intacts pour chaque écran, avec le bon type (entier, pas une chaîne),
    faute de quoi une comparaison stricte "===" côté JS échouerait
    silencieusement."""
    event_id = db.create_custom_event(game, "NEED_ACTION")

    listener_screen = _create_screen(client, game, "Écran écouteur")
    listener_node_id = _add_flow_node(
        client,
        game,
        listener_screen,
        node_type="trigger",
        trigger_event="evenement",
        trigger_custom_event_id=event_id,
    )

    dispatcher_screen = _create_screen(client, game, "Écran déclencheur")
    dispatcher_node_id = _add_flow_node(
        client,
        game,
        dispatcher_screen,
        node_type="action",
        action_type="declencher_evenement",
        target_custom_event_id=event_id,
    )

    payload = game_engine.full_game_payload(game)
    flows = payload["flows"]

    listener_node = next(n for n in flows[str(listener_screen)]["nodes"] if n["id"] == listener_node_id)
    assert listener_node["trigger_event"] == "evenement"
    assert listener_node["trigger_custom_event_id"] == event_id

    dispatcher_node = next(n for n in flows[str(dispatcher_screen)]["nodes"] if n["id"] == dispatcher_node_id)
    assert dispatcher_node["action_type"] == "declencher_evenement"
    assert dispatcher_node["target_custom_event_id"] == event_id