test(integration): 3 scénarios E2E Playwright (Fiches, Analyse, Plan d'action)

- Infrastructure test : StreamlitApp helper, réinitialisation session, serveur dédié port 8502
- Scénario Fiches : sélection dossier Assemblage + fiche casques VR
- Scénario Analyse : Serveur→Pays géo, Germanium, Chine, filtre IHH Pays, vérification Sankey
- Scénario Plan d'action : Serveur + Germanium, vérification dashboard criticités

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

tests/integration/conftest.py

@@ -0,0 +1,130 @@
+"""Fixtures pour les tests d'intégration Playwright + Streamlit."""
+
+import json
+import re
+import subprocess
+import sys
+import time
+from pathlib import Path
+from urllib.request import urlopen
+
+import pytest
+
+pytest.importorskip("playwright")
+
+TEST_PORT = 8502
+PROJECT_ROOT = Path(__file__).parent.parent.parent
+SESSION_DIR = PROJECT_ROOT / "tmp" / "sessions" / "anonymous"
+STATUT_FILE = SESSION_DIR / "statut_general.json"
+
+# État vierge pour garantir des tests reproductibles
+ETAT_VIERGE = {
+    "navigation_onglet": "Instructions",
+    "theme_mode": "Clair",
+    "pages": {},
+}
+
+
+class StreamlitApp:
+    """Encapsule les interactions avec les widgets Streamlit via Playwright."""
+
+    RERUN_WAIT = 2000  # ms à attendre après chaque interaction widget
+
+    def __init__(self, page):
+        """Initialise avec une page Playwright."""
+        self.page = page
+
+    def naviguer_vers(self, onglet):
+        """Clique sur un onglet du menu latéral."""
+        self.page.get_by_role("button", name=onglet).click()
+        self.page.wait_for_timeout(self.RERUN_WAIT)
+
+    def choisir_selectbox(self, label, valeur):
+        """Sélectionne une valeur dans un selectbox Streamlit."""
+        combobox = self.page.get_by_role("combobox", name=re.compile(label))
+        combobox.click()
+        self.page.get_by_role("option", name=valeur).click()
+        self.page.wait_for_timeout(self.RERUN_WAIT)
+
+    def ajouter_multiselect(self, label, valeur):
+        """Ajoute une valeur à un multiselect Streamlit en filtrant par texte."""
+        combobox = self.page.get_by_role("combobox", name=re.compile(label))
+        combobox.click()
+        self.page.wait_for_timeout(500)
+        self.page.keyboard.type(valeur, delay=50)
+        self.page.wait_for_timeout(1000)
+        self.page.get_by_role("option", name=valeur).click()
+        # Fermer le dropdown pour éviter qu'il n'intercepte les clics suivants
+        self.page.keyboard.press("Escape")
+        self.page.wait_for_timeout(self.RERUN_WAIT)
+
+    def cocher(self, label):
+        """Coche une checkbox Streamlit via le label texte (l'input natif est caché)."""
+        texte = self.page.get_by_text(re.compile(label)).first
+        texte.scroll_into_view_if_needed()
+        texte.click()
+        self.page.wait_for_timeout(self.RERUN_WAIT)
+
+    def choisir_radio(self, valeur):
+        """Sélectionne une option radio Streamlit via le label texte (l'input natif est caché)."""
+        # Cibler le <p> dans le conteneur du radiogroup
+        radio_label = self.page.locator(f'[role="radiogroup"] p:text-is("{valeur}")')
+        radio_label.scroll_into_view_if_needed()
+        radio_label.click()
+        self.page.wait_for_timeout(self.RERUN_WAIT)
+
+    def cliquer_bouton(self, label):
+        """Clique sur un bouton Streamlit."""
+        self.page.get_by_role("button", name=re.compile(label)).click()
+        self.page.wait_for_timeout(self.RERUN_WAIT)
+
+
+def _reinitialiser_session():
+    """Remet le fichier de session à un état vierge."""
+    SESSION_DIR.mkdir(parents=True, exist_ok=True)
+    STATUT_FILE.write_text(json.dumps(ETAT_VIERGE, ensure_ascii=False, indent=2))
+
+
+@pytest.fixture(scope="session")
+def streamlit_server():
+    """Démarre un serveur Streamlit dédié aux tests d'intégration."""
+    _reinitialiser_session()
+    url = f"http://localhost:{TEST_PORT}"
+    process = subprocess.Popen(
+        [
+            sys.executable, "-m", "streamlit", "run", "fabnum.py",
+            "--server.port", str(TEST_PORT),
+            "--server.headless", "true",
+            "--browser.gatherUsageStats", "false",
+        ],
+        cwd=str(PROJECT_ROOT),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+    for _ in range(30):
+        try:
+            response = urlopen(url)
+            if response.status == 200:
+                break
+        except OSError:
+            pass
+        time.sleep(1)
+    else:
+        process.terminate()
+        pytest.fail("Le serveur Streamlit n'a pas démarré en 30 secondes")
+
+    yield url
+
+    process.terminate()
+    process.wait(timeout=10)
+
+
+@pytest.fixture
+def app(page, streamlit_server):
+    """Application Streamlit prête à interagir (état réinitialisé)."""
+    _reinitialiser_session()
+    page.set_viewport_size({"width": 1280, "height": 900})
+    page.goto(streamlit_server)
+    page.wait_for_timeout(3000)
+    return StreamlitApp(page)

tests/integration/test_analyse.py

@@ -0,0 +1,48 @@
+"""Test E2E – Scénario 2 : Analyse complète Produit final → Pays géographique."""
+
+import pytest
+from playwright.sync_api import expect
+
+pytestmark = pytest.mark.integration
+
+
+def test_analyse_serveur_germanium_chine(app):
+    """Formulaire complet : Serveur → Pays géo, Germanium, Chine, filtre IHH Pays."""
+    page = app.page
+
+    # Naviguer vers l'onglet Analyse (le graphe se charge automatiquement)
+    app.naviguer_vers("Analyse")
+    expect(page.get_by_role("heading", name="Analyse du graphe")).to_be_visible(timeout=15000)
+
+    # Sélectionner le niveau de départ : Produit final
+    app.choisir_selectbox("Niveau de départ", "Produit final")
+
+    # Sélectionner le niveau d'arrivée : Pays géographique
+    app.choisir_selectbox("Niveau d'arrivée", "Pays géographique")
+
+    # Ajouter le minerai Germanium
+    expect(page.get_by_text("Filtrer par minerais")).to_be_visible(timeout=10000)
+    app.ajouter_multiselect("minerais", "Germanium")
+
+    # Ajouter le nœud de départ : Serveur
+    app.ajouter_multiselect("noeuds de départ", "Serveur")
+
+    # Ajouter le nœud d'arrivée : Chine
+    app.ajouter_multiselect("noeuds d'arrivée", "Chine")
+
+    # Cocher le filtre IHH
+    app.cocher("IHH")
+
+    # Sélectionner Pays pour le filtre IHH
+    app.choisir_radio("Pays")
+
+    # Lancer l'analyse
+    app.cliquer_bouton("Lancer l'analyse")
+
+    # Vérifier que le diagramme Sankey s'affiche
+    expect(page.get_by_text("Hiérarchie filtrée")).to_be_visible(timeout=15000)
+
+    # Vérifier la présence de nœuds clés dans le graphe Sankey (Plotly)
+    sankey = page.get_by_test_id("stPlotlyChart")
+    expect(sankey.get_by_text("Serveur").first).to_be_visible()
+    expect(sankey.get_by_text("Germanium").first).to_be_visible()

tests/integration/test_fiches.py

@@ -0,0 +1,37 @@
+"""Test E2E – Scénario 1 : Navigation et affichage des fiches."""
+
+import pytest
+from playwright.sync_api import expect
+
+pytestmark = pytest.mark.integration
+
+
+def test_selectionner_dossier_et_fiche(app):
+    """Sélectionne Assemblage puis la première fiche et vérifie le contenu."""
+    page = app.page
+
+    # Naviguer vers l'onglet Fiches
+    app.naviguer_vers("Fiches")
+    expect(page.get_by_role("heading", name="Découverte des fiches")).to_be_visible(timeout=10000)
+
+    # Sélectionner le dossier Assemblage
+    app.choisir_selectbox("catégorie de fiches", "Assemblage")
+    expect(page.get_by_text("Choisissez une fiche")).to_be_visible(timeout=10000)
+
+    # Sélectionner la première fiche (pattern unique pour éviter l'ambiguïté avec "catégorie de fiches")
+    app.choisir_selectbox("Choisissez une fiche", "Fiche assemblage casques VR.md")
+
+    # Vérifier le titre de la fiche
+    expect(page.get_by_role("heading", name="Fiche assemblage Casque VR")).to_be_visible(timeout=10000)
+
+    # Vérifier les sections principales (les titres apparaissent dans <summary> et <h2>,
+    # on utilise .first pour éviter la strict mode violation)
+    expect(page.get_by_text("Présentation synthétique").first).to_be_visible()
+    expect(page.get_by_text("Composants assemblés").first).to_be_visible()
+    expect(page.get_by_text("Principaux assembleurs").first).to_be_visible()
+
+    # Vérifier le tableau de version
+    expect(page.get_by_text("Version initiale")).to_be_visible()
+
+    # Vérifier la section tickets
+    expect(page.get_by_text("Gestion des tickets")).to_be_visible()

tests/integration/test_plan_action.py

@@ -0,0 +1,35 @@
+"""Test E2E – Scénario 3 : Plan d'action Serveur + Germanium."""
+
+import pytest
+from playwright.sync_api import expect
+
+pytestmark = pytest.mark.integration
+
+
+def test_plan_action_serveur_germanium(app):
+    """Soumet Serveur + Germanium et vérifie le dashboard de résultats."""
+    page = app.page
+
+    # Naviguer vers l'onglet Plan d'action
+    app.naviguer_vers("Plan d'action")
+    expect(page.get_by_role("heading", name="Analyse du graphe pour action")).to_be_visible(timeout=15000)
+
+    # Sélectionner le produit final : Serveur
+    app.ajouter_multiselect("noeuds de départ", "Serveur")
+
+    # Ajouter le minerai : Germanium
+    app.ajouter_multiselect("minerais", "Germanium")
+
+    # Soumettre la demande
+    app.cliquer_bouton("Soumettre")
+
+    # Vérifier le dashboard (stage 1)
+    expect(page.get_by_text("Panneau de sélection")).to_be_visible(timeout=30000)
+    expect(page.get_by_text("Top chaînes critiques")).to_be_visible()
+
+    # Vérifier la chaîne critique principale
+    expect(page.get_by_text("Serveur ↔ Carte mère ↔ Germanium").first).to_be_visible()
+
+    # Vérifier les criticités
+    expect(page.get_by_text("Synthèse des criticités")).to_be_visible()
+    expect(page.get_by_text("Criticité globale").first).to_be_visible()