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)."""
import json
import re

import db
import screens


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


def _add_element(client, slug, screen_id, widget="bouton"):
    resp = client.post(f"/game/{slug}/screens/{screen_id}/elements/add", data={"widget": widget}, follow_redirects=False)
    return int(re.search(r"selected=(\d+)", resp.headers["Location"]).group(1))


def _add_flow_node(client, slug, screen_id, **fields):
    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, game):
    screen_id = _create_screen(client, game)
    resp = client.post(f"/game/{game}/events/new", data={
        "name": "NEED_ACTION", "description": "Besoin d'action", "has_element_param": "on", "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"
    assert events[0]["has_element_param"] == 1

    # 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_and_param_flag(client, game):
    screen_id = _create_screen(client, game)
    event_id = db.create_custom_event(game, "NEED_ACTION", "old", False)
    resp = client.post(f"/game/{game}/events/{event_id}/edit", data={
        "description": "new", "has_element_param": "on", "screen_id": screen_id,
    })
    assert resp.status_code == 302
    event = db.get_custom_event(game, event_id)
    assert event["description"] == "new"
    assert event["has_element_param"] == 1


def test_usages_reports_listener_and_dispatcher_across_different_screens(client, game):
    event_id = db.create_custom_event(game, "NEED_ACTION", "", True)

    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")
    element_id = _add_element(client, game, dispatcher_screen)
    _add_flow_node(
        client, game, dispatcher_screen,
        node_type="action", action_type="declencher_evenement",
        target_custom_event_id=event_id, target_element_id=element_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, game):
    event_id = db.create_custom_event(game, "NEED_ACTION", "", True)

    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,
    )

    # Le modèle (élément de jeu réutilisable) est un écran caché — voir
    # screens.create_element_type — le déclencheur y est posé exactement
    # comme sur une vraie scène.
    screens.create_element_type(game, "Carte email", "📧")
    template = next(t for t in screens.list_element_types(game) if t["name"] == "Carte email")
    template_screen_id = template["template_screen_id"]
    dispatcher_node_id = _add_flow_node(
        client, game, template_screen_id,
        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/modèles 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, template_screen_id) is not None


def test_runtime_payload_exposes_both_nodes_for_the_client_js(client, game):
    """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", "", True)

    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")
    element_id = _add_element(client, game, dispatcher_screen)
    dispatcher_node_id = _add_flow_node(
        client, game, dispatcher_screen,
        node_type="action", action_type="declencher_evenement",
        target_custom_event_id=event_id, target_element_id=element_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
    assert dispatcher_node["target_element_id"] == element_id


def test_modifier_element_from_event_flag_persists(client, game):
    screen_id = _create_screen(client, game)
    node_id = _add_flow_node(
        client, game, screen_id,
        node_type="action", action_type="modifier_element",
        target_element_from_event=1, element_property="surbrillance", element_value="pulse",
    )
    node = screens.get_flow_node(game, node_id)
    assert node["target_element_from_event"] == 1
    assert node["target_element_id"] is None


def test_flow_node_run_data_resolves_event_row_id_from_request_body(client, game):
    """Miroir serveur du sentinel EVENT_ROW_ID (voir play.html,
    readFieldValue) — flow_node_run_data.py doit résoudre target_row_id
    depuis le corps JSON ("event_row_id") exactement comme il le fait déjà
    pour CLICKED_ROW_ID ("clicked_row_id")."""
    resp = client.post(f"/game/{game}/objects/new", data={
        "object_name": "Jauge", "field_name[]": ["valeur"],
        "field_type[]": ["nombre_entier"], "field_relation[]": [""], "field_required[]": ["0"],
        "field_min[]": [""], "field_max[]": [""],
    }, follow_redirects=False)
    def_id = int(resp.headers["Location"].rstrip("/").split("/")[-1])
    client.post(f"/game/{game}/objects/{def_id}/data/new", data={"valeur": "1"})
    client.post(f"/game/{game}/objects/{def_id}/data/new", data={"valeur": "2"})
    rows = db.list_rows(game, db.get_definition(game, def_id))
    row_id = next(r["id"] for r in rows if r["valeur"] == 2)

    screen_id = _create_screen(client, game)
    node_id = _add_flow_node(
        client, game, screen_id,
        node_type="action", action_type="modifier_donnee",
        target_definition_id=def_id, target_row_id=screens.EVENT_ROW_ID,
        target_field="valeur", data_operation="definir_nombre", data_value="99",
    )

    resp = client.post(
        f"/game/{game}/flow/nodes/{node_id}/run-data",
        data=json.dumps({"clicked_row_id": None, "event_row_id": row_id}),
        content_type="application/json",
    )
    assert resp.status_code == 200
    assert resp.get_json()["ok"] is True

    row = db.get_row(game, db.get_definition(game, def_id), row_id)
    assert row["valeur"] == 99