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 bannerActive = () => document.getElementById("offline-banner").classList.contains("active");
const inst = () => FakeWebSocket.instances;

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());
  FakeWebSocket.autoOpen = false;
  inst()[0]._serverClose();
}, 250);

// KLUCZOWY test regresji: po zamknięciu MUSI powstać nowe połączenie.
setTimeout(() => {
  assert("6. reconnect zaplanowany i wykonany", inst().length >= 2);
}, 5000);

setTimeout(() => {
  assert("7. baner offline widoczny po 10 s", bannerActive());
  FakeWebSocket.autoOpen = true;
  inst()[inst().length-1]._open();
  inst()[inst().length-1]._message({type:"machine_data", machine:"4.09", data:{status:"Alarm - test"}});
}, 16000);

setTimeout(() => {
  assert("8. po powrocie: baner znika, alarm 4.09", !bannerActive() && cards().join()==="4.09");
}, 17000);

setTimeout(() => { window.__przedWatchdog = inst().length; }, 18000);
setTimeout(() => {
  assert("9. watchdog: cisza 100s wymusza nowe polaczenie", inst().length > window.__przedWatchdog);
}, 125000);

setTimeout(() => {
  assert("10. przeterminowany alarm znika z ekranu", !alarmActive());
  document.getElementById("wynik").textContent += "\\nKONIEC";
}, 340000);
"""

FAKE_WS = """
class FakeWebSocket {
  static instances = [];
  static autoOpen = true;
  constructor(url) {
    this.url = url;
    this.readyState = 0;
    FakeWebSocket.instances.push(this);
    if (FakeWebSocket.autoOpen) setTimeout(() => this._open(), 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-banner{{display:none}} #offline-banner.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=350000",
                "--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())