515 lines
22 KiB
Python
515 lines
22 KiB
Python
#!/usr/bin/env python3
|
||
"""Exercise the persisted Sleep Diary projection incrementally in Firefox."""
|
||
|
||
import json
|
||
import os
|
||
import signal
|
||
import socket
|
||
import subprocess
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
BUILD = ROOT / os.environ.get("TRAINLOG_WEB_E2E_BUILD", "build") / "tui"
|
||
EXECUTABLE = Path(os.environ.get("TRAINLOG_WEB_E2E_EXECUTABLE", BUILD / "trainlog"))
|
||
GECKO = Path(
|
||
os.environ.get(
|
||
"TRAINLOG_GECKODRIVER", "/home/fy59/.cache/trainlog/web-e2e-tools/geckodriver"
|
||
)
|
||
)
|
||
EVIDENCE = ROOT / "docs" / "reviews" / "evidence"
|
||
|
||
|
||
class BrowserSleepDiaryTest(unittest.TestCase):
|
||
def test_live_agenda_uses_each_persisted_revision_without_reload(self):
|
||
if not GECKO.is_file():
|
||
self.skipTest("pinned geckodriver unavailable")
|
||
|
||
try:
|
||
from selenium import webdriver
|
||
from selenium.webdriver.common.by import By
|
||
from selenium.webdriver.firefox.service import Service
|
||
from selenium.webdriver.support.ui import Select, WebDriverWait
|
||
import websocket
|
||
except ModuleNotFoundError:
|
||
self.skipTest("selenium unavailable")
|
||
|
||
with tempfile.TemporaryDirectory(prefix="trainlog-web-sleep-") as directory:
|
||
root = Path(directory)
|
||
environment = os.environ.copy()
|
||
environment.update(
|
||
{
|
||
"XDG_DATA_HOME": str(root / "data"),
|
||
"XDG_CONFIG_HOME": str(root / "config"),
|
||
"XDG_CACHE_HOME": str(root / "cache"),
|
||
}
|
||
)
|
||
configured_port = os.environ.get("TRAINLOG_WEB_E2E_PORT")
|
||
if configured_port is None:
|
||
with socket.socket() as reserved:
|
||
reserved.bind(("127.0.0.1", 0))
|
||
port = reserved.getsockname()[1]
|
||
else:
|
||
port = int(configured_port)
|
||
server = subprocess.Popen(
|
||
[str(EXECUTABLE), "--web", "--port", str(port)],
|
||
env=environment,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
start_new_session=True,
|
||
)
|
||
driver = 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)
|
||
|
||
EVIDENCE.mkdir(parents=True, exist_ok=True)
|
||
options = webdriver.FirefoxOptions()
|
||
options.add_argument("-headless")
|
||
options.set_capability("webSocketUrl", True)
|
||
options.set_preference("browser.download.folderList", 2)
|
||
options.set_preference("browser.download.dir", str(EVIDENCE))
|
||
options.set_preference("browser.download.useDownloadDir", True)
|
||
options.set_preference(
|
||
"browser.helperApps.neverAsk.saveToDisk", "application/pdf"
|
||
)
|
||
options.set_preference("pdfjs.disabled", True)
|
||
driver = webdriver.Firefox(options=options, service=Service(str(GECKO)))
|
||
driver.set_window_size(1440, 1100)
|
||
wait = WebDriverWait(driver, 15)
|
||
base_url = os.environ.get(
|
||
"TRAINLOG_WEB_E2E_BASE_URL", f"http://127.0.0.1:{port}"
|
||
)
|
||
driver.get(f"{base_url}/analyse?section=sleep")
|
||
wait.until(
|
||
lambda current: current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-workspace']"
|
||
)
|
||
)
|
||
driver.execute_script(
|
||
"""
|
||
window.__trainlogErrors = [];
|
||
window.__trainlogFetches = [];
|
||
addEventListener('error', event => window.__trainlogErrors.push(String(event.error || event.message)));
|
||
addEventListener('unhandledrejection', event => window.__trainlogErrors.push(String(event.reason)));
|
||
const originalFetch = window.fetch;
|
||
window.fetch = async (...args) => {
|
||
try {
|
||
const response = await originalFetch(...args);
|
||
window.__trainlogFetches.push({url: String(args[0]), method: args[1]?.method || 'GET', status: response.status});
|
||
return response;
|
||
} catch (error) {
|
||
window.__trainlogErrors.push(String(error));
|
||
throw error;
|
||
}
|
||
};
|
||
"""
|
||
)
|
||
|
||
def set_value(test_id: str, value: str):
|
||
element = driver.find_element(
|
||
By.CSS_SELECTOR, f"[data-testid='{test_id}']"
|
||
)
|
||
driver.execute_script(
|
||
"Object.getOwnPropertyDescriptor(HTMLInputElement.prototype,'value').set.call(arguments[0],arguments[1]);"
|
||
"arguments[0].dispatchEvent(new Event('input',{bubbles:true}));"
|
||
"arguments[0].dispatchEvent(new Event('change',{bubbles:true}));",
|
||
element,
|
||
value,
|
||
)
|
||
|
||
def click(element):
|
||
driver.execute_script(
|
||
"arguments[0].scrollIntoView({block:'center'}); arguments[0].click();",
|
||
element,
|
||
)
|
||
|
||
def post_count() -> int:
|
||
return driver.execute_script(
|
||
"return window.__trainlogFetches.filter(item => item.method === 'POST').length"
|
||
)
|
||
|
||
def await_persisted(previous_posts: int):
|
||
wait.until(lambda current: post_count() > previous_posts)
|
||
wait.until(
|
||
lambda current: current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-save-state']"
|
||
).text
|
||
== "Enregistré localement"
|
||
)
|
||
|
||
def add_medication(dose: str):
|
||
click(driver.find_element(
|
||
By.XPATH,
|
||
"//button[normalize-space()='+ Ajouter un médicament']",
|
||
))
|
||
set_value("sleep-medication-name", "venlafaxine")
|
||
set_value("sleep-medication-dose", dose)
|
||
before = post_count()
|
||
click(driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-add-medication']"
|
||
))
|
||
wait.until(lambda current: post_count() > before)
|
||
wait.until(
|
||
lambda current: not current.find_elements(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-medication-name']"
|
||
)
|
||
)
|
||
|
||
add_medication("75")
|
||
add_medication("37.5")
|
||
medication_select = Select(
|
||
driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-intake-medication']"
|
||
)
|
||
)
|
||
option_labels = [option.text for option in medication_select.options]
|
||
self.assertIn("venlafaxine — 75 mg", option_labels)
|
||
self.assertIn("venlafaxine — 37,5 mg", option_labels)
|
||
|
||
for label, dose in (
|
||
("venlafaxine — 75 mg", "75"),
|
||
("venlafaxine — 37,5 mg", "37.5"),
|
||
):
|
||
medication_select.select_by_visible_text(label)
|
||
wait.until(
|
||
lambda current: current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-intake-dose']"
|
||
).get_attribute("value")
|
||
== dose
|
||
)
|
||
before = post_count()
|
||
click(driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-add-intake']"
|
||
))
|
||
await_persisted(before)
|
||
wait.until(
|
||
lambda current: current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-agenda-medication']"
|
||
).text
|
||
== "M ×2"
|
||
)
|
||
|
||
def add_event(kind: str, start: str, end: str | None, assertion):
|
||
Select(
|
||
driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-event-type']"
|
||
)
|
||
).select_by_value(kind)
|
||
set_value("sleep-event-start", start)
|
||
if end is not None:
|
||
set_value("sleep-event-end", end)
|
||
before = post_count()
|
||
click(driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-add-event']"
|
||
))
|
||
await_persisted(before)
|
||
wait.until(assertion)
|
||
|
||
add_event(
|
||
"bed_time",
|
||
"22:45",
|
||
None,
|
||
lambda current: current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-agenda-bed-time']"
|
||
).text.endswith("22:45"),
|
||
)
|
||
add_event(
|
||
"sleep",
|
||
"23:15",
|
||
"03:00",
|
||
lambda current: "3 h 45 min"
|
||
in current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-agenda-sleep-duration']"
|
||
).text,
|
||
)
|
||
add_event(
|
||
"long_awake",
|
||
"03:00",
|
||
"03:30",
|
||
lambda current: "1 · 30 min"
|
||
in current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-agenda-long-awake']"
|
||
).text,
|
||
)
|
||
add_event(
|
||
"sleep",
|
||
"03:30",
|
||
"06:45",
|
||
lambda current: "7 h 00 min"
|
||
in current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-agenda-sleep-duration']"
|
||
).text,
|
||
)
|
||
add_event(
|
||
"final_get_up",
|
||
"07:10",
|
||
None,
|
||
lambda current: "8 h 25 min"
|
||
in current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-agenda-time-in-bed']"
|
||
).text,
|
||
)
|
||
|
||
expected = driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid^='sleep-agenda-row-']"
|
||
).text
|
||
mutation_errors = driver.execute_script(
|
||
"return window.__trainlogErrors"
|
||
)
|
||
mutation_fetches = driver.execute_script(
|
||
"return window.__trainlogFetches"
|
||
)
|
||
self.assertEqual([], mutation_errors)
|
||
self.assertTrue(mutation_fetches)
|
||
self.assertTrue(
|
||
all(item["status"] == 200 for item in mutation_fetches)
|
||
)
|
||
driver.refresh()
|
||
wait.until(
|
||
lambda current: "8 h 25 min"
|
||
in current.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-agenda-time-in-bed']"
|
||
).text
|
||
)
|
||
actual = driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid^='sleep-agenda-row-']"
|
||
).text
|
||
self.assertEqual(expected, actual)
|
||
self.assertTrue(
|
||
driver.find_element(
|
||
By.CSS_SELECTOR, "[data-testid='sleep-workspace']"
|
||
).is_displayed()
|
||
)
|
||
self.assertLessEqual(
|
||
driver.execute_script("return document.documentElement.scrollWidth"),
|
||
driver.execute_script("return document.documentElement.clientWidth"),
|
||
)
|
||
bidi = websocket.create_connection(
|
||
driver.capabilities["webSocketUrl"], suppress_origin=True
|
||
)
|
||
bidi.send(
|
||
json.dumps(
|
||
{"id": 1, "method": "browsingContext.getTree", "params": {}}
|
||
)
|
||
)
|
||
tree = json.loads(bidi.recv())
|
||
context = tree["result"]["contexts"][0]["context"]
|
||
bidi.send(
|
||
json.dumps(
|
||
{
|
||
"id": 2,
|
||
"method": "browsingContext.setViewport",
|
||
"params": {
|
||
"context": context,
|
||
"viewport": {"width": 390, "height": 844},
|
||
"devicePixelRatio": 1,
|
||
},
|
||
}
|
||
)
|
||
)
|
||
response = json.loads(bidi.recv())
|
||
self.assertEqual("success", response["type"])
|
||
wait.until(lambda current: current.execute_script("return innerWidth") == 390)
|
||
self.assertLessEqual(
|
||
driver.execute_script("return document.documentElement.scrollWidth"),
|
||
driver.execute_script("return document.documentElement.clientWidth"),
|
||
)
|
||
bidi.send(
|
||
json.dumps(
|
||
{
|
||
"id": 3,
|
||
"method": "browsingContext.setViewport",
|
||
"params": {
|
||
"context": context,
|
||
"viewport": {"width": 1440, "height": 1100},
|
||
"devicePixelRatio": 1,
|
||
},
|
||
}
|
||
)
|
||
)
|
||
self.assertEqual("success", json.loads(bidi.recv())["type"])
|
||
self.assertTrue(
|
||
driver.save_screenshot(
|
||
str(EVIDENCE / "sleep-diary-live-agenda-firefox.png")
|
||
)
|
||
)
|
||
|
||
def stable_id(prefix: str) -> str:
|
||
return f"{prefix}_{uuid.uuid4()}"
|
||
|
||
extra_entries = []
|
||
for day in range(14, 20):
|
||
start_date = f"2026-09-{day:02d}"
|
||
end_date = f"2026-09-{day + 1:02d}"
|
||
|
||
def at(clock: str) -> str:
|
||
date = start_date if int(clock[:2]) >= 18 else end_date
|
||
return f"{date}T{clock}:00+02:00"
|
||
|
||
extra_entries.append(
|
||
{
|
||
"entry_id": "",
|
||
"expected_revision": None,
|
||
"night_start_date": start_date,
|
||
"night_end_date": end_date,
|
||
"created_at": at("18:00"),
|
||
"updated_at": at("18:00"),
|
||
"sleep_quality": "B",
|
||
"wake_quality": "Moy",
|
||
"day_form": "TB",
|
||
"treatment_and_notes": "Remarque française synthétique.",
|
||
"events": [
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "bed_time",
|
||
"start_at": at("22:30"),
|
||
"end_at": None,
|
||
},
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "sleep",
|
||
"start_at": at("23:00"),
|
||
"end_at": at("03:00"),
|
||
},
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "long_awake",
|
||
"start_at": at("03:00"),
|
||
"end_at": at("03:45"),
|
||
},
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "sleep",
|
||
"start_at": at("03:45"),
|
||
"end_at": at("07:00"),
|
||
},
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "half_sleep",
|
||
"start_at": at("07:00"),
|
||
"end_at": at("07:10"),
|
||
},
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "final_get_up",
|
||
"start_at": at("07:15"),
|
||
"end_at": None,
|
||
},
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "nap",
|
||
"start_at": at("14:00"),
|
||
"end_at": at("14:35"),
|
||
},
|
||
{
|
||
"event_id": stable_id("sle"),
|
||
"type": "daytime_sleepiness",
|
||
"start_at": at("15:00"),
|
||
"end_at": None,
|
||
},
|
||
],
|
||
"intakes": [],
|
||
}
|
||
)
|
||
responses = driver.execute_async_script(
|
||
"""
|
||
const entries=arguments[0], done=arguments[arguments.length-1];
|
||
(async () => {
|
||
const status = await fetch('/api/v1/sync/status');
|
||
const token = status.headers.get('X-Trainlog-CSRF-Token');
|
||
const responses = [];
|
||
for (const entry of entries) {
|
||
const response = await fetch('/api/v1/sleep-diary', {
|
||
method: 'POST',
|
||
headers: {'Accept':'application/json','Content-Type':'application/json','X-Trainlog-CSRF-Token':token},
|
||
body: JSON.stringify(entry),
|
||
});
|
||
responses.push({status: response.status, body: await response.json()});
|
||
}
|
||
done(responses);
|
||
})().catch(error => done([{status:0, body:String(error)}]));
|
||
""",
|
||
extra_entries,
|
||
)
|
||
self.assertEqual([200] * 6, [item["status"] for item in responses])
|
||
driver.refresh()
|
||
wait.until(
|
||
lambda current: len(
|
||
current.find_elements(
|
||
By.CSS_SELECTOR, "[data-testid^='sleep-agenda-row-']"
|
||
)
|
||
)
|
||
== 7
|
||
)
|
||
EVIDENCE.mkdir(parents=True, exist_ok=True)
|
||
download_path = EVIDENCE / "trainlog-sleep-diary.pdf"
|
||
pdf_path = EVIDENCE / "sleep-diary-fr-7-days.pdf"
|
||
raster_prefix = EVIDENCE / "sleep-diary-fr-7-days"
|
||
raster_path = EVIDENCE / "sleep-diary-fr-7-days.png"
|
||
numbered_download = EVIDENCE / "trainlog-sleep-diary(1).pdf"
|
||
for artifact in (
|
||
download_path,
|
||
numbered_download,
|
||
pdf_path,
|
||
raster_path,
|
||
):
|
||
if artifact.exists():
|
||
artifact.unlink()
|
||
click(driver.find_element(By.XPATH, "//button[normalize-space()='Exporter PDF']"))
|
||
wait.until(
|
||
lambda current: download_path.is_file()
|
||
and download_path.stat().st_size > 0
|
||
)
|
||
download_path.replace(pdf_path)
|
||
extracted = root / "sleep-diary-fr-7-days.txt"
|
||
subprocess.run(
|
||
["pdftotext", str(pdf_path), str(extracted)], check=True
|
||
)
|
||
pdf_text = extracted.read_text(encoding="utf-8")
|
||
self.assertIn("QUALITÉ DU", pdf_text)
|
||
self.assertIn("LONG RÉVEIL", pdf_text.upper())
|
||
self.assertIn("AVERTISSEMENT : 7 jours non validés", pdf_text)
|
||
for forbidden in (
|
||
"WAKE",
|
||
"TREATMENT",
|
||
"AWAKE",
|
||
"sleepiness",
|
||
"medication intake",
|
||
"nights",
|
||
):
|
||
self.assertNotIn(forbidden, pdf_text)
|
||
subprocess.run(
|
||
[
|
||
"pdftoppm",
|
||
"-png",
|
||
"-f",
|
||
"1",
|
||
"-singlefile",
|
||
"-r",
|
||
"150",
|
||
str(pdf_path),
|
||
str(raster_prefix),
|
||
],
|
||
check=True,
|
||
)
|
||
bidi.close()
|
||
finally:
|
||
if driver is not None:
|
||
driver.quit()
|
||
try:
|
||
os.killpg(server.pid, signal.SIGTERM)
|
||
except ProcessLookupError:
|
||
pass
|
||
server.wait(timeout=10)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|