Forge-Engine/tests/test_custom_events.py

"""Événements personnalisés (db/custom_events/, screens/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 screens/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 screens
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 = screens.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 screens.get_flow_node(game, listener_node_id) is None
    assert screens.get_flow_node(game, dispatcher_node_id) is None
    # Les écrans eux-mêmes ne sont pas touchés, seuls les nœuds.
    assert screens.get_screen(game, listener_screen) is not None
    assert screens.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/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 = screens.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