daynivo-kiosk-os/tests/test_agent.py

import json

import pytest

from voxip_kiosk.agent import Agent, HUNG_RELOAD_AFTER, HUNG_RESTART_AFTER
from voxip_kiosk.cdp import CDPError
from voxip_kiosk.config import save_config


class Clock:
    def __init__(self):
        self.t = 1000.0

    def __call__(self):
        return self.t


class BrowserStub:
    def __init__(self, url):
        self.url = url
        self.navigations = []
        self.reloads = 0
        self.closed = False
        self.responsive_answer = True
        self.down = False

    def main_page(self):
        if self.down:
            raise CDPError("down")
        return {"url": self.url, "webSocketDebuggerUrl": "ws://127.0.0.1:9222/devtools/page/1"}

    def new_page(self, url):
        self.url = url

    def navigate(self, url):
        self.navigations.append(url)
        self.url = url

    def reload(self, ignore_cache=False):
        self.reloads += 1

    def responsive(self, ws=None, timeout=5):
        return self.responsive_answer

    def close_browser(self):
        self.closed = True


class Probe:
    def __init__(self):
        self.ok = True
        self.reason = ""
        self.calls = 0

    def __call__(self, url):
        self.calls += 1
        return self.ok, self.reason


@pytest.fixture
def env(paths):
    paths.admin_file.write_text("{}")  # setup finished
    clock = Clock()
    probe = Probe()
    start = f"file://{paths.www_dir}/start.html?next=x"
    browser = BrowserStub(start)
    agent = Agent(paths, cdp=browser, probe_fn=probe, clock=clock, wall=clock, hostname=lambda: "kiosk",
                  addresses=lambda: {"eth0": ["192.168.1.20/24"], "wlan0": ["fe80::1/64"]})
    return agent, browser, probe, clock, paths


def test_navigates_from_start_page_when_reachable(env):
    agent, browser, probe, clock, paths = env
    agent.tick()
    assert browser.navigations == ["https://schermen-ijsselflat.voxip.nl"]
    state = json.loads(paths.agent_state_file.read_text())
    assert state["page"] == "start" and state["target_reachable"] is True


def test_offline_page_when_unreachable_then_recovers(env):
    agent, browser, probe, clock, _ = env
    probe.ok, probe.reason = False, "Naam niet gevonden (DNS)"
    agent.tick()
    assert "offline.html" in browser.url and "reason=Naam+niet+gevonden" in browser.url
    navs = len(browser.navigations)
    clock.t += 5
    agent.tick()  # still offline: no re-navigation (probe cached / page already offline)
    assert len(browser.navigations) == navs
    probe.ok = True
    clock.t += 11
    agent.tick()
    assert browser.url == "https://schermen-ijsselflat.voxip.nl"


def test_chromium_error_page_is_replaced(env):
    agent, browser, probe, clock, _ = env
    browser.url = "chrome-error://chromewebdata/"
    probe.ok = False
    agent.tick()
    assert "offline.html" in browser.url


def test_running_page_is_left_alone_when_the_server_is_unreachable(env):
    """Pages with their own offline cache (e.g. InfoFlow displays) must keep running."""
    agent, browser, probe, clock, _ = env
    browser.url = "https://schermen-ijsselflat.voxip.nl"
    agent.tick()
    probe.ok = False
    for _ in range(20):
        clock.t += 61
        agent.tick()
    assert browser.navigations == []
    assert browser.reloads == 0


def test_url_change_navigates_without_restart(env):
    agent, browser, probe, clock, paths = env
    browser.url = "https://schermen-ijsselflat.voxip.nl"
    agent.tick()
    save_config(paths.config_file, {"url": "https://nieuw.example.nl/scherm"})
    agent.tick()
    assert browser.navigations == ["https://nieuw.example.nl/scherm"]
    assert not browser.closed


def test_refresh_interval_skips_when_unreachable(env):
    agent, browser, probe, clock, paths = env
    save_config(paths.config_file, {"refresh_interval": 60})
    browser.url = "https://schermen-ijsselflat.voxip.nl"
    agent.tick()
    clock.t += 61
    agent.tick()
    assert browser.reloads == 1
    probe.ok = False
    clock.t += 61
    agent.tick()
    assert browser.reloads == 1
    assert "overgeslagen" in agent.state["last_error"]["message"]


def test_hung_renderer_reload_then_browser_restart(env):
    agent, browser, probe, clock, _ = env
    browser.url = "https://schermen-ijsselflat.voxip.nl"
    browser.responsive_answer = False
    for _ in range(HUNG_RELOAD_AFTER):
        agent.tick()
    assert browser.reloads == 1 and not browser.closed
    for _ in range(HUNG_RESTART_AFTER - HUNG_RELOAD_AFTER):
        agent.tick()
    assert browser.closed


def test_setup_page_until_password_exists(env):
    agent, browser, probe, clock, paths = env
    paths.admin_file.unlink()
    paths.setup_code_file.write_text("ABCD-EFGH\n")
    agent.tick()
    assert "setup.html" in browser.url
    assert "code=ABCD-EFGH" in browser.url and "ips=192.168.1.20" in browser.url and "port=8080" in browser.url
    agent.tick()
    assert len(browser.navigations) == 1  # no flicker: setup page is not reloaded every tick
    paths.admin_file.write_text("{}")
    agent.tick()
    assert browser.url == "https://schermen-ijsselflat.voxip.nl"


def test_devtools_down_is_reported(env):
    agent, browser, probe, clock, paths = env
    browser.down = True
    agent.tick()
    assert json.loads(paths.agent_state_file.read_text())["browser"] == "unreachable"


def test_offline_page_never_receives_the_token(env, paths):
    agent, browser, probe, _, _ = env
    save_config(paths.config_file, {"url": "https://e.org/scherm?token=geheim123456"})
    probe.ok, probe.reason = False, "Time-out"
    agent.tick()
    assert "offline.html" in browser.url
    assert "geheim" not in browser.url and "3456" in browser.url


def test_offline_page_when_the_network_link_is_gone_then_back(env):
    agent, browser, probe, clock, _ = env
    network = {"eth0": ["192.168.1.20/24"]}
    agent.addresses = lambda: dict(network)
    browser.url = "https://schermen-ijsselflat.voxip.nl"
    agent.tick()
    assert browser.navigations == []

    network.clear()  # cable out
    probe.ok = False
    clock.t += 5
    agent.tick()
    assert "offline.html" in browser.url and "netwerkkabel" in browser.url
    clock.t += 5
    agent.tick()
    assert len(browser.navigations) == 1  # no re-navigation while it stays down

    network["eth0"] = ["192.168.1.20/24"]  # cable back in
    probe.ok = True
    clock.t += 11
    agent.tick()
    assert browser.url == "https://schermen-ijsselflat.voxip.nl"