Forge-Engine/tests/test_flow_blocks.py

"""Blocs de logique (screens/flow/blocks/) : regroupe les nœuds d'UN écran
en sous-graphes nommés/décrits pour rester lisible même à l'échelle de
milliers de nœuds (voir templates/screen_edit.html, onglet "🧩 Blocs de
logique"). Un bloc est un automate FERMÉ : aucune arête possible entre deux
nœuds de blocs différents (toute communication inter-blocs passe par les
événements personnalisés, voir tests/test_custom_events.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_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 _add_flow_edge(client, slug, screen_id, from_node_id, to_node_id, from_port="out"):
    return client.post(
        f"/game/{slug}/screens/{screen_id}/flow/edges/add",
        data=json.dumps({"from_node_id": from_node_id, "from_port": from_port, "to_node_id": to_node_id}),
        content_type="application/json",
    )


def test_create_and_list_flow_blocks(client, game):
    screen_id = _create_screen(client, game)
    resp = client.post(f"/game/{game}/screens/{screen_id}/flow/blocks/new", data={
        "name": "Ouvrir la porte", "description": "Ouvre la porte au clic.",
    })
    assert resp.status_code == 302
    blocks = screens.list_flow_blocks(game, screen_id)
    assert len(blocks) == 1
    assert blocks[0]["name"] == "Ouvrir la porte"


def test_orphan_nodes_are_migrated_into_a_default_block(client, game):
    screen_id = _create_screen(client, game)
    # Nœud posé "avant" l'existence des blocs (aucun block_id transmis) —
    # doit être rattaché à un "Bloc principal" auto-créé au premier
    # appel de list_flow_blocks, sans rien perdre.
    node_id = _add_flow_node(client, game, screen_id, node_type="trigger", trigger_event="affichage")

    blocks = screens.list_flow_blocks(game, screen_id)
    assert len(blocks) == 1
    assert blocks[0]["name"] == "Bloc principal"
    node = screens.get_flow_node(game, node_id)
    assert node["block_id"] == blocks[0]["id"]

    # Idempotent : un second appel ne crée pas un second "Bloc principal".
    blocks_again = screens.list_flow_blocks(game, screen_id)
    assert len(blocks_again) == 1


def test_node_created_with_a_block_id_is_not_touched_by_migration(client, game):
    screen_id = _create_screen(client, game)
    block_id = screens.create_flow_block(game, screen_id, "Mon bloc")
    node_id = _add_flow_node(
        client, game, screen_id,
        node_type="trigger", trigger_event="affichage", block_id=block_id,
    )
    blocks = screens.list_flow_blocks(game, screen_id)
    assert len(blocks) == 1
    assert blocks[0]["id"] == block_id
    assert screens.get_flow_node(game, node_id)["block_id"] == block_id


def test_edge_between_two_different_blocks_is_rejected(client, game):
    screen_id = _create_screen(client, game)
    block_a = screens.create_flow_block(game, screen_id, "Bloc A")
    block_b = screens.create_flow_block(game, screen_id, "Bloc B")
    node_a = _add_flow_node(client, game, screen_id, node_type="trigger", trigger_event="affichage", block_id=block_a)
    node_b = _add_flow_node(client, game, screen_id, node_type="action", action_type="attendre", data_value="1", block_id=block_b)

    resp = _add_flow_edge(client, game, screen_id, node_a, node_b)
    assert resp.status_code == 400

    # Une arête entre deux nœuds du MÊME bloc, elle, doit réussir.
    node_a2 = _add_flow_node(client, game, screen_id, node_type="action", action_type="attendre", data_value="1", block_id=block_a)
    resp_ok = _add_flow_edge(client, game, screen_id, node_a, node_a2)
    assert resp_ok.status_code == 200


def test_deleting_a_block_cascades_only_its_own_nodes_and_edges(client, game):
    screen_id = _create_screen(client, game)
    block_a = screens.create_flow_block(game, screen_id, "Bloc A")
    block_b = screens.create_flow_block(game, screen_id, "Bloc B")

    node_a1 = _add_flow_node(client, game, screen_id, node_type="trigger", trigger_event="affichage", block_id=block_a)
    node_a2 = _add_flow_node(client, game, screen_id, node_type="action", action_type="attendre", data_value="1", block_id=block_a)
    edge_resp = _add_flow_edge(client, game, screen_id, node_a1, node_a2)
    assert edge_resp.status_code == 200
    edge_id = edge_resp.get_json()["id"]

    node_b1 = _add_flow_node(client, game, screen_id, node_type="trigger", trigger_event="affichage", block_id=block_b)

    resp = client.post(f"/game/{game}/screens/{screen_id}/flow/blocks/{block_a}/delete", data={}, follow_redirects=False)
    assert resp.status_code == 302

    assert screens.get_flow_block(game, block_a) is None
    assert screens.get_flow_node(game, node_a1) is None
    assert screens.get_flow_node(game, node_a2) is None
    assert screens.list_flow_edges(game, screen_id) == [] or all(e["id"] != edge_id for e in screens.list_flow_edges(game, screen_id))

    # Le bloc B et son nœud, eux, ne sont pas touchés.
    assert screens.get_flow_block(game, block_b) is not None
    assert screens.get_flow_node(game, node_b1) is not None


def test_screen_edit_page_shows_correct_node_count_right_after_migration(client, game):
    """routes/screens/screen_edit.py doit lancer list_flow_blocks() (qui
    migre les nœuds orphelins vers "Bloc principal") AVANT list_flow_nodes()
    — sinon la page affiche 0 nœud pour "Bloc principal" au tout premier
    chargement suivant la migration (block_id périmé en mémoire)."""
    screen_id = _create_screen(client, game)
    _add_flow_node(client, game, screen_id, node_type="trigger", trigger_event="affichage")
    resp = client.get(f"/game/{game}/screens/{screen_id}/edit")
    assert resp.status_code == 200
    assert b"Bloc principal" in resp.data
    # Le compte de nœuds (colonne "Nœuds") doit valoir 1, pas 0 — cherche
    # la cellule <td>1</td> qui suit le nom du bloc dans le tableau.
    html = resp.get_data(as_text=True)
    row = html[html.index("Bloc principal"):]
    assert re.search(r"<td>1</td>", row)


def test_screen_edit_page_renders_blocks_tab(client, game):
    screen_id = _create_screen(client, game)
    screens.create_flow_block(game, screen_id, "Ouvrir la porte", "Ouvre la porte. " * 5)
    resp = client.get(f"/game/{game}/screens/{screen_id}/edit")
    assert resp.status_code == 200
    assert b"Blocs de logique" in resp.data
    assert b"Ouvrir la porte" in resp.data