#!/usr/bin/env python3 """Exercise Web Sessions through embedded production assets in real Firefox.""" import json import os import signal import socket import sqlite3 import subprocess import sys import tempfile import time import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[1] BUILD = ROOT / os.environ.get("TRAINLOG_WEB_E2E_BUILD", "build") / "tui" GECKO = Path( os.environ.get( "TRAINLOG_GECKODRIVER", "/home/fy59/.cache/trainlog/web-e2e-tools/geckodriver" ) ) EVIDENCE = ROOT / "docs" / "reviews" / "evidence" class BrowserSessionsTest(unittest.TestCase): def test_programs_v1_import_detail_and_discrete_delete_in_real_firefox(self): if not GECKO.is_file(): self.skipTest("pinned geckodriver unavailable") from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.common.keys import Keys from selenium.webdriver.firefox.service import Service from selenium.webdriver.support.ui import WebDriverWait import websocket with tempfile.TemporaryDirectory(prefix="trainlog-web-programs-") as directory: root = Path(directory) xdg = root / "data" config = root / "config" xdg.mkdir() config.mkdir() with socket.socket() as reserved: reserved.bind(("127.0.0.1", 0)) port = reserved.getsockname()[1] environment = os.environ.copy() environment.update( { "XDG_DATA_HOME": str(xdg), "XDG_CONFIG_HOME": str(config), "XDG_CACHE_HOME": str(root / "cache"), } ) server = subprocess.Popen( [str(BUILD / "trainlog"), "--web", "--port", str(port)], env=environment, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) driver = None bidi = None try: for _ in range(200): with socket.socket() as probe: if probe.connect_ex(("127.0.0.1", port)) == 0: break time.sleep(0.025) options = webdriver.FirefoxOptions() options.add_argument("-headless") options.set_capability("webSocketUrl", True) driver = webdriver.Firefox(options=options, service=Service(str(GECKO))) wait = WebDriverWait(driver, 15) bidi = websocket.create_connection( driver.capabilities["webSocketUrl"], suppress_origin=True ) command_id = 0 def bidi_command(method: str, parameters: dict): nonlocal command_id command_id += 1 bidi.send(json.dumps({ "id": command_id, "method": method, "params": parameters, })) while True: response = json.loads(bidi.recv()) if response.get("id") != command_id: continue if response.get("type") != "success": raise AssertionError(response) return response["result"] context = bidi_command("browsingContext.getTree", {})["contexts"][0]["context"] def set_viewport(width: int, height: int): bidi_command( "browsingContext.setViewport", { "context": context, "viewport": {"width": width, "height": height}, "devicePixelRatio": 1, }, ) wait.until( lambda current: current.execute_script( "return innerWidth===arguments[0]&&innerHeight===arguments[1]", width, height, ) ) set_viewport(1440, 1000) driver.get(f"http://127.0.0.1:{port}/seances") wait.until(lambda current: "Aucun résultat" in current.page_source) catalog = driver.execute_async_script( "const done=arguments[0];" "fetch('/api/v1/sessions/catalog?offset=0&limit=1')" ".then((response)=>response.json()).then(done).catch(String);" ) self.assertIsInstance(catalog, dict) self.assertTrue(catalog["items"]) exercise = catalog["items"][0] occurrence = { "entry_id": "pge_cccccccc-cccc-4ccc-8ccc-cccccccccccc", "exercise_id": exercise["exercise_id"], "equipment_id": None, "load_mode": "none", "rest_seconds": 90, "target_sets": 3 if exercise["recording_mode"] == "sets" else None, "target_reps": 8 if exercise["tracking_mode"] == "reps" else None, "target_duration_seconds": ( None if exercise["tracking_mode"] == "reps" else 600 ), "target_weight_kg": None, "notes": None, } program = { "format": "trainlog-program", "version": 1, "program": { "program_id": "pg_aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", "title": "Cycle navigateur V1", "note": "Programme synthétique lisible.", "state": "active", "start_date": "2026-09-21", "end_date": None, "sessions": [ { "program_session_id": "pgs_bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", "title": "Séance A ordonnée", "session_type": "training", "planned_for": "2026-09-21", "note": None, "occurrences": [occurrence], }, { "program_session_id": "pgs_dddddddd-dddd-4ddd-8ddd-dddddddddddd", "title": "Séance B ordonnée", "session_type": "max_test", "planned_for": "2026-09-23", "note": "Contrôle.", "occurrences": [{**occurrence, "entry_id": "pge_eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"}], }, ], }, } import_path = root / "programme-navigateur-v1.json" import_path.write_text(json.dumps(program), encoding="utf-8") driver.find_element(By.XPATH, "//button[normalize-space()='Programmes']").click() wait.until(lambda current: "Importer un programme" in current.page_source) driver.find_element(By.XPATH, "//button[normalize-space()='Importer un programme']").click() driver.find_element(By.CSS_SELECTOR, "input[type='file']").send_keys(str(import_path)) wait.until(lambda current: "Aperçu avant import" in current.page_source) self.assertIn("2 séances", driver.page_source) driver.find_element(By.XPATH, "//button[normalize-space()='Importer']").click() wait.until(lambda current: "Cycle navigateur V1" in current.page_source) wait.until(lambda current: "Séance A ordonnée" in current.page_source) detail = driver.find_element(By.CSS_SELECTOR, "main").text self.assertIn("PROGRAMME · ACTIF", detail) self.assertIn("Import trainlog-program v1", detail) self.assertLess(detail.index("1. Séance A ordonnée"), detail.index("2. Séance B ordonnée")) driver.find_element( By.XPATH, "//button[normalize-space()='← Retour aux programmes']" ).click() wait.until(lambda current: "Programme actif" in current.page_source) card = driver.find_element(By.CSS_SELECTOR, ".program-row").text self.assertIn("Cycle navigateur V1", card) self.assertIn("2 séances · jamais utilisé", card) self.assertIn("trainlog-program · importé", card) EVIDENCE.mkdir(parents=True, exist_ok=True) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-programs-v1-card-desktop.png") )) trash = driver.find_element( By.CSS_SELECTOR, "button[aria-label='Supprimer le programme Cycle navigateur V1']" ) trash.click() confirmation = wait.until( lambda current: current.find_element(By.CSS_SELECTOR, "[role='alertdialog']") ) self.assertIn("Supprimer le programme « Cycle navigateur V1 » ?", confirmation.text) self.assertIn("séances sources", confirmation.text) self.assertNotIn("Séance A ordonnée", confirmation.text) set_viewport(390, 844) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-programs-v1-delete-dialog-mobile-390x844.png") )) confirmation.find_element(By.XPATH, ".//button[normalize-space()='Conserver']").click() wait.until(lambda current: not current.find_elements(By.CSS_SELECTOR, "[role='alertdialog']")) self.assertEqual(driver.switch_to.active_element.get_attribute("aria-label"), trash.get_attribute("aria-label")) trash = driver.find_element( By.CSS_SELECTOR, "button[aria-label='Supprimer le programme Cycle navigateur V1']" ) trash.click() wait.until(lambda current: current.find_element(By.CSS_SELECTOR, "[role='alertdialog']")) driver.switch_to.active_element.send_keys(Keys.ESCAPE) wait.until(lambda current: not current.find_elements(By.CSS_SELECTOR, "[role='alertdialog']")) database_path = xdg / "trainlog" / "trainlog.db" with sqlite3.connect(database_path) as database: self.assertIsNone(database.execute( "SELECT deleted_at FROM programs WHERE program_id=?", (program["program"]["program_id"],), ).fetchone()[0]) driver.find_element( By.CSS_SELECTOR, "button[aria-label='Supprimer le programme Cycle navigateur V1']" ).click() confirmation = wait.until( lambda current: current.find_element(By.CSS_SELECTOR, "[role='alertdialog']") ) confirmation.find_element( By.XPATH, ".//button[normalize-space()='Supprimer le programme']" ).click() wait.until( lambda current: "Aucun programme." in current.page_source or current.find_elements(By.CSS_SELECTOR, "[role='alert']") ) self.assertNotIn("Cycle navigateur V1", driver.find_element(By.CSS_SELECTOR, "main").text) driver.refresh() wait.until(lambda current: "Cycle navigateur V1" not in current.page_source) with sqlite3.connect(database_path) as database: deleted_at = database.execute( "SELECT deleted_at FROM programs WHERE program_id=?", (program["program"]["program_id"],), ).fetchone()[0] source_sessions = database.execute( "SELECT COUNT(*) FROM program_sessions WHERE program_id=?", (program["program"]["program_id"],), ).fetchone()[0] source_entries = database.execute( "SELECT COUNT(*) FROM program_session_entries" ).fetchone()[0] self.assertIsNotNone(deleted_at) self.assertEqual(source_sessions, 2) self.assertEqual(source_entries, 2) finally: if bidi is not None: bidi.close() if driver is not None: driver.quit() os.killpg(server.pid, signal.SIGTERM) server.wait(timeout=10) def test_create_preferences_reload_and_withdraw_by_real_click(self): if not GECKO.is_file(): self.skipTest("pinned geckodriver unavailable") from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.firefox.service import Service from selenium.webdriver.support.ui import WebDriverWait import websocket with tempfile.TemporaryDirectory(prefix="trainlog-web-sessions-") as directory: root = Path(directory) xdg = root / "data" config = root / "config" xdg.mkdir() config.mkdir() with socket.socket() as reserved: reserved.bind(("127.0.0.1", 0)) port = reserved.getsockname()[1] environment = os.environ.copy() environment.update( { "XDG_DATA_HOME": str(xdg), "XDG_CONFIG_HOME": str(config), "XDG_CACHE_HOME": str(root / "cache"), } ) server = subprocess.Popen( [str(BUILD / "trainlog"), "--web", "--port", str(port)], env=environment, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, ) driver = None bidi = None try: for _ in range(200): with socket.socket() as probe: if probe.connect_ex(("127.0.0.1", port)) == 0: break time.sleep(0.025) options = webdriver.FirefoxOptions() options.add_argument("-headless") options.set_capability("webSocketUrl", True) driver = webdriver.Firefox(options=options, service=Service(str(GECKO))) wait = WebDriverWait(driver, 15) bidi = websocket.create_connection( driver.capabilities["webSocketUrl"], suppress_origin=True ) command_id = 0 def bidi_command(method: str, parameters: dict): nonlocal command_id command_id += 1 bidi.send(json.dumps({ "id": command_id, "method": method, "params": parameters, })) while True: response = json.loads(bidi.recv()) if response.get("id") != command_id: continue if response.get("type") != "success": raise AssertionError(response) return response["result"] contexts = bidi_command("browsingContext.getTree", {})["contexts"] self.assertEqual(len(contexts), 1) context = contexts[0]["context"] def set_viewport(width: int, height: int): bidi_command( "browsingContext.setViewport", { "context": context, "viewport": {"width": width, "height": height}, "devicePixelRatio": 1, }, ) wait.until( lambda current: current.execute_script( "return innerWidth===arguments[0]&&innerHeight===arguments[1]", width, height, ) ) set_viewport(1440, 1000) driver.get(f"http://127.0.0.1:{port}/seances") wait.until(lambda current: "Aucun résultat" in current.page_source) EVIDENCE.mkdir(parents=True, exist_ok=True) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-sessions-list-desktop.png") )) driver.find_element(By.XPATH, "//button[normalize-space()='Programmes']").click() wait.until(lambda current: "Importer un programme" in current.page_source) set_viewport(390, 844) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-programs-empty-mobile.png") )) driver.find_element(By.XPATH, "//button[normalize-space()='Préparation']").click() wait.until(lambda current: "Nouvelle séance" in current.page_source) driver.find_element(By.XPATH, "//button[normalize-space()='Nouvelle séance']").click() wait.until(lambda current: "Nouvelle séance" in current.page_source) title = driver.find_element(By.XPATH, "//label[normalize-space()='Titre']/input") title.send_keys("Séance navigateur éphémère") planned = driver.find_element( By.XPATH, "//label[normalize-space()='Date planifiée']/input" ) driver.execute_script( "const setter=Object.getOwnPropertyDescriptor(" "HTMLInputElement.prototype,'value').set;" "setter.call(arguments[0],'2026-09-18');" "arguments[0].dispatchEvent(new Event('input',{bubbles:true}));" "arguments[0].dispatchEvent(new Event('change',{bubbles:true}));", planned, ) catalog_choice = wait.until( lambda current: current.find_element( By.CSS_SELECTOR, ".catalog-results button" ) ) catalog_choice.click() set_viewport(390, 844) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-sessions-editor-mobile.png") )) save = driver.find_element( By.XPATH, "//button[normalize-space()='Marquer prête / Préparer pour Android']", ) driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", save) save.click() wait.until(lambda current: "PRÉPARATION MANUELLE" in current.page_source) driver.refresh() wait.until(lambda current: "Séance navigateur éphémère" in current.page_source) set_viewport(1440, 1000) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-sessions-detail-desktop.png") )) self.assertIn("18/09/2026", driver.page_source) driver.find_element( By.XPATH, "//button[normalize-space()='Paramètres d’affichage']" ).click() wait.until(lambda current: "Format des dates" in current.page_source) driver.find_element(By.CSS_SELECTOR, "input[value='iso']").click() wait.until( lambda current: current.find_element( By.CSS_SELECTOR, "main time" ).text == "2026-09-18" ) driver.refresh() wait.until( lambda current: current.find_element( By.CSS_SELECTOR, "main time" ).text == "2026-09-18" ) preferences = ( config / "trainlog" / "web" / "preferences-v1.json" ).read_text(encoding="utf-8") self.assertIn('"date_format":"iso"', preferences) driver.find_element( By.XPATH, "//button[normalize-space()='Supprimer la préparation']" ).click() confirmation = wait.until( lambda current: current.find_element(By.CSS_SELECTOR, "[role='alertdialog']") ) self.assertIn("Séance navigateur éphémère", confirmation.text) set_viewport(390, 844) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-sessions-delete-confirmation-mobile.png") )) confirmation.find_element( By.XPATH, ".//button[normalize-space()='Supprimer la préparation']" ).click() wait.until( lambda current: "annulation Android à synchroniser" in current.page_source ) set_viewport(1440, 1000) self.assertTrue(driver.save_screenshot( str(EVIDENCE / "web-sessions-withdrawn-detail-desktop.png") )) driver.refresh() wait.until(lambda current: "Supprimée" in current.page_source) self.assertNotIn( "Supprimer la préparation", driver.find_element(By.CSS_SELECTOR, "main").text, ) database_path = xdg / "trainlog" / "trainlog.db" with sqlite3.connect(database_path) as database: row = database.execute( "SELECT r.title, p.editing_state, p.withdrawn_at " "FROM session_preparations p " "JOIN session_preparation_revisions r " "ON r.revision_id = p.current_revision_id" ).fetchone() entry_count = database.execute( "SELECT COUNT(*) FROM session_preparation_entries" ).fetchone()[0] withdrawal_count = database.execute( "SELECT COUNT(*) FROM session_preparation_withdrawals" ).fetchone()[0] delivery_count = database.execute( "SELECT COUNT(*) FROM session_preparation_deliveries" ).fetchone()[0] self.assertEqual(row[:2], ("Séance navigateur éphémère", "ready")) self.assertIsNotNone(row[2]) self.assertEqual(entry_count, 1) self.assertEqual(delivery_count, 1) self.assertEqual(withdrawal_count, 1) export_path = root / "session-preparations-v2.json" subprocess.run( [ sys.executable, str(ROOT / "tools" / "export_session_preparations.py"), str(export_path), "--database", str(database_path), ], check=True, capture_output=True, text=True, ) exported = json.loads(export_path.read_text(encoding="utf-8")) self.assertEqual(exported["deliveries"], []) self.assertEqual(len(exported["withdrawals"]), 1) self.assertEqual(len(exported["withdrawals"][0]["deliveries"]), 1) finally: if bidi is not None: bidi.close() if driver is not None: driver.quit() os.killpg(server.pid, signal.SIGTERM) server.wait(timeout=10) if __name__ == "__main__": unittest.main()