Code/tests/integration/conftest.py

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