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
from typing import Any

from flask.testing import FlaskClient

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 _add_flow_edge(
    client: FlaskClient, slug: str, screen_id: int, from_node_id: int, to_node_id: int, from_port: str = "out"
) -> Any:
    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: FlaskClient, game: str) -> None:
    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: FlaskClient, game: str) -> None:
    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 = not_none(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: FlaskClient, game: str) -> None:
    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 not_none(screens.get_flow_node(game, node_id))["block_id"] == block_id


def test_edge_between_two_different_blocks_is_rejected(client: FlaskClient, game: str) -> None:
    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: FlaskClient, game: str) -> None:
    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_loads_after_orphan_node_migration(client: FlaskClient, game: str) -> None:
    """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). Porté sur
    jeu_2d (voir routes/scenes/scene_edit_view.py, blocks_view) — l'onglet
    "Blocs de logique" lui-même n'existe que dans l'éditeur document
    retiré ; on vérifie ici directement l'état migré plutôt que son rendu."""
    screen_id = _create_screen(client, game)
    node_id = _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
    blocks = screens.list_flow_blocks(game, screen_id)
    assert len(blocks) == 1
    assert blocks[0]["name"] == "Bloc principal"
    assert not_none(screens.get_flow_node(game, node_id))["block_id"] == blocks[0]["id"]