als-alert-screen/test/testy_ekranu.py

#!/usr/bin/env python3
"""Testy ekranu alarmowego — deterministyczne, bez sieci i bez serwera.

Skrypt aplikacji jest wycinany z ŻYWEGO als_alert_screen.html (nie z kopii,
więc testy nie mogą się rozjechać z kodem), a `window.WebSocket` podmieniany
na fałszywkę sterowaną scenariuszem. Cały przebieg — łącznie z 90-sekundowym
watchdogiem i 5-minutowym wygasaniem danych — jedzie na wirtualnym czasie
Chromium (`--virtual-time-budget`), więc trwa sekundy.

Uruchomienie:
    python3 test/testy_ekranu.py
    CHROME_BIN=/usr/bin/chromium python3 test/testy_ekranu.py

Kod wyjścia 0 = wszystkie scenariusze PASS.

Dlaczego to istnieje: poprzednia wersja strony miała błąd, przez który KAŻDA
maszyna po pierwszym zerwaniu połączenia milkła na zawsze (wyjątek w onclose
przerywał handler przed zaplanowaniem reconnectu — patrz README, Historia).
Objaw „alarmy przestały wyskakiwać" był leczony restartem z crona co 2 h.
Scenariusz 6 jest bezpośrednim testem regresji tego błędu.
"""

from __future__ import annotations

import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

APP = Path(__file__).resolve().parent.parent / "als_alert_screen.html"

SCENARIUSZ = """
const wyniki = [];
function assert(nazwa, warunek) {
  wyniki.push((warunek ? "PASS " : "FAIL ") + nazwa);
  document.getElementById("wynik").textContent = wyniki.join("\\n");
}
const alarmActive = () => document.getElementById("alarm-view").classList.contains("active");
const cards = () => [...document.querySelectorAll(".card-machine-id")].map(e => e.textContent);
const offline = () => document.getElementById("offline-indicator").classList.contains("active");
const inst = () => FakeWebSocket.instances;
const otwarty = () => inst().find(w => w.readyState === 1);

setTimeout(() => {
  assert("1. polaczenie nawiazane", inst().length === 1 && inst()[0].readyState === 1);
  inst()[0]._message({type:"machine_data", machine:"2.60", data:{status:"Alarm - awaria"}});
  inst()[0]._message({type:"machine_data", machine:"3.22", data:{status:"Automatyczny"}});
}, 50);

setTimeout(() => {
  assert("2. migawka: widok alarmowy z 2.60", alarmActive() && cards().join()==="2.60");
  inst()[0]._message({type:"machine_update", machine:"3.22", data:{status:"Alarm - przekroczony cykl"}});
}, 100);

setTimeout(() => {
  assert("3. update na zywo: dwie karty", alarmActive() && cards().join()==="2.60,3.22");
  inst()[0]._message({type:"machine_update", machine:"2.60", data:{status:"Automatyczny"}});
  inst()[0]._message({type:"machine_update", machine:"3.22", data:{status:"Automatyczny"}});
}, 150);

setTimeout(() => {
  assert("4. koniec alarmow: powrot do iframe", !alarmActive());
  inst()[0]._message({type:"machine_update", machine:"9.99", data:{status:"Alarm - obca"}});
}, 200);

setTimeout(() => {
  assert("5. maszyna spoza filtra ignorowana", !alarmActive());
  // serwer pada NA DLUGO: kazda kolejna proba polaczenia jest odrzucana
  FakeWebSocket.autoOpen = false;
  inst()[0]._serverClose();
}, 250);

// KLUCZOWY test regresji starego bledu: po zamknieciu MUSI powstac nowa proba.
setTimeout(() => {
  assert("6. reconnect zaplanowany i wykonany", inst().length >= 2);
}, 5000);

// ANTY-MIGANIE (regresja 2026-07-31): przy DLUGIEJ przerwie wskaznik ma
// swiecic CIAGLE. Blad resetowal disconnectedSince przy kazdej nieudanej
// probie, wiec przy odstepie ponowien > 10 s wskaznik cyklicznie znikal.
// Probkujemy co 500 ms przez 90 s ciaglej awarii — z bledem ~30% probek
// bylo "zgaszonych", po poprawce wszystkie musza byc aktywne.
window.__probki = { aktywne: 0, wszystkie: 0 };
const probkowanie = setInterval(() => {
  window.__probki.wszystkie++;
  if (offline()) window.__probki.aktywne++;
}, 500);
setTimeout(() => { window.__probki.aktywne = 0; window.__probki.wszystkie = 0; }, 20000);
setTimeout(() => {
  clearInterval(probkowanie);
  const p = window.__probki;
  assert("7. wskaznik offline swieci CIAGLE przez cala przerwe (" +
    p.aktywne + "/" + p.wszystkie + " probek)", p.wszystkie > 100 && p.aktywne === p.wszystkie);
  // serwer wraca — najblizsza zaplanowana proba wstanie
  FakeWebSocket.autoOpen = true;
}, 111000);

setTimeout(() => {
  assert("8. po powrocie serwera: polaczono, wskaznik zgaszony", !!otwarty() && !offline());
  otwarty() && otwarty()._message({type:"machine_data", machine:"4.09", data:{status:"Alarm - test"}});
}, 155000);

setTimeout(() => {
  assert("9. alarm po powrocie widoczny", alarmActive() && cards().join()==="4.09");
  window.__przedWatchdog = inst().length;
}, 156000);

// cisza > 90 s -> watchdog ubija polaczenie, reconnect wstaje od reki
setTimeout(() => {
  assert("10. watchdog: cisza wymusza nowe polaczenie", inst().length > window.__przedWatchdog);
}, 270000);

// dane 4.09 z t=155 s bez odswiezenia -> po 5 min znikaja z ekranu
setTimeout(() => {
  assert("11. przeterminowany alarm znika z ekranu", !alarmActive());
  document.getElementById("wynik").textContent += "\\nKONIEC";
}, 470000);
"""

FAKE_WS = """
class FakeWebSocket {
  static instances = [];
  // true  = polaczenie wstaje (serwer dziala)
  // false = polaczenie ODRZUCANE po chwili (serwer lezy / zly adres) —
  //         przegladarka w tej sytuacji wola onclose, wiec fake tez musi,
  //         inaczej nie da sie przetestowac dlugiej przerwy z ponowieniami.
  static autoOpen = true;
  constructor(url) {
    this.url = url;
    this.readyState = 0;
    FakeWebSocket.instances.push(this);
    if (FakeWebSocket.autoOpen) setTimeout(() => this._open(), 5);
    else setTimeout(() => this._serverClose(), 5);
  }
  _open() {
    if (this.readyState !== 0) return;
    this.readyState = 1;
    this.onopen && this.onopen();
  }
  _message(obj) {
    this.onmessage && this.onmessage({ data: JSON.stringify(obj) });
  }
  _serverClose() {
    if (this.readyState === 3) return;
    this.readyState = 3;
    this.onclose && this.onclose();
  }
  close() {
    if (this.readyState === 3) return;
    this.readyState = 3;
    setTimeout(() => this.onclose && this.onclose(), 1);
  }
}
FakeWebSocket.OPEN = 1;
window.WebSocket = FakeWebSocket;
window.WebSocket.OPEN = 1;
"""


def zbuduj_harness() -> str:
    app = APP.read_text(encoding="utf-8")
    skrypt = re.search(r"<script>([\s\S]*)</script>", app).group(1)
    markup = re.search(r"<body>([\s\S]*?)<script>", app).group(1)
    # iframe bez realnego adresu — test działa offline
    markup = re.sub(r'src="http://[^"]*"', 'src="about:blank"', markup)
    return f"""<!doctype html><html><head><meta charset="utf-8"><style>
#alarm-view{{display:none}} #alarm-view.active{{display:flex}}
#offline-indicator{{display:none}} #offline-indicator.active{{display:block}}
</style></head><body>
{markup}
<pre id="wynik"></pre>
<script>{FAKE_WS}</script>
<script>{skrypt}</script>
<script>{SCENARIUSZ}</script>
</body></html>"""


def znajdz_chromium() -> str:
    for kandydat in (
        os.environ.get("CHROME_BIN"),
        shutil.which("chromium"),
        shutil.which("chromium-browser"),
        shutil.which("google-chrome"),
        "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
    ):
        if kandydat and Path(kandydat).exists():
            return kandydat
    sys.exit("Nie znaleziono Chromium/Chrome — ustaw CHROME_BIN.")


def main() -> int:
    chrome = znajdz_chromium()
    with tempfile.TemporaryDirectory() as tmp:
        harness = Path(tmp) / "harness.html"
        harness.write_text(zbuduj_harness(), encoding="utf-8")
        wynik = subprocess.run(
            [
                chrome,
                "--headless",
                "--disable-gpu",
                "--no-sandbox",
                "--virtual-time-budget=480000",
                "--dump-dom",
                f"file://{harness}",
            ],
            capture_output=True,
            text=True,
            timeout=180,
        )
    m = re.search(r'<pre id="wynik">([\s\S]*?)</pre>', wynik.stdout)
    if not m:
        print("BŁĄD: brak wyników w DOM — harness nie wystartował?")
        print(wynik.stdout[:500])
        return 2
    linie = [l for l in m.group(1).strip().splitlines() if l]
    for linia in linie:
        print(linia)
    if "KONIEC" not in linie:
        print("BŁĄD: scenariusz nie dobiegł końca (za mały virtual-time-budget?)")
        return 2
    return 1 if any(l.startswith("FAIL") for l in linie) else 0


if __name__ == "__main__":
    sys.exit(main())