109 lines
3.0 KiB
Python
109 lines
3.0 KiB
Python
"""Dashboard (status + control) tests for the web server. Stdlib only."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
import urllib.request
|
|
|
|
import pytest
|
|
|
|
from gowelcome.web import MjpegServer
|
|
|
|
_FAKE_JPEG = bytes.fromhex("ffd8ffe000104a46494600010100000100010000ffd9")
|
|
|
|
|
|
@pytest.fixture()
|
|
def dash():
|
|
state = {"play_mode": "moderate", "paused": False}
|
|
|
|
def status():
|
|
return {"state": "WANDER", "play_mode": state["play_mode"],
|
|
"paused": state["paused"]}
|
|
|
|
def control(cmd):
|
|
action = cmd.get("action")
|
|
if action == "play_mode":
|
|
state["play_mode"] = cmd.get("mode")
|
|
return {"ok": True, "play_mode": state["play_mode"]}
|
|
if action == "pause":
|
|
state["paused"] = True
|
|
return {"ok": True}
|
|
return {"ok": False, "error": "unknown"}
|
|
|
|
srv = MjpegServer(lambda: _FAKE_JPEG, host="127.0.0.1", port=0, fps=30,
|
|
status_provider=status, control_handler=control)
|
|
srv.start()
|
|
time.sleep(0.2)
|
|
try:
|
|
yield srv, state
|
|
finally:
|
|
srv.stop()
|
|
|
|
|
|
def _get(port, path):
|
|
r = urllib.request.urlopen(f"http://127.0.0.1:{port}{path}", timeout=3)
|
|
return r.status, r.read()
|
|
|
|
|
|
def _post(port, path, obj):
|
|
data = json.dumps(obj).encode()
|
|
req = urllib.request.Request(f"http://127.0.0.1:{port}{path}", data=data,
|
|
headers={"Content-Type": "application/json"})
|
|
r = urllib.request.urlopen(req, timeout=3)
|
|
return r.status, json.loads(r.read())
|
|
|
|
|
|
def test_index_has_controls(dash):
|
|
srv, _ = dash
|
|
assert srv.has_controls
|
|
_s, body = _get(srv.port, "/")
|
|
assert b"control dashboard" in body
|
|
assert b"/control" in body and b"E-STOP" in body
|
|
|
|
|
|
def test_status_json(dash):
|
|
srv, _ = dash
|
|
_s, body = _get(srv.port, "/status.json")
|
|
obj = json.loads(body)
|
|
assert obj["state"] == "WANDER"
|
|
assert obj["play_mode"] == "moderate"
|
|
|
|
|
|
def test_post_control_play_mode(dash):
|
|
srv, state = dash
|
|
status, resp = _post(srv.port, "/control", {"action": "play_mode", "mode": "playful"})
|
|
assert status == 200 and resp["ok"] is True
|
|
assert state["play_mode"] == "playful"
|
|
# reflected in subsequent status
|
|
_s, body = _get(srv.port, "/status.json")
|
|
assert json.loads(body)["play_mode"] == "playful"
|
|
|
|
|
|
def test_post_control_pause(dash):
|
|
srv, state = dash
|
|
_s, resp = _post(srv.port, "/control", {"action": "pause"})
|
|
assert resp["ok"] is True
|
|
assert state["paused"] is True
|
|
|
|
|
|
def test_post_unknown_action(dash):
|
|
srv, _ = dash
|
|
_s, resp = _post(srv.port, "/control", {"action": "nope"})
|
|
assert resp["ok"] is False
|
|
|
|
|
|
def test_viewer_only_has_no_controls():
|
|
srv = MjpegServer(lambda: _FAKE_JPEG, host="127.0.0.1", port=0)
|
|
srv.start()
|
|
time.sleep(0.2)
|
|
try:
|
|
assert not srv.has_controls
|
|
_s, body = _get(srv.port, "/")
|
|
assert b"live camera" in body
|
|
# control endpoints disabled
|
|
_s2, resp = _post(srv.port, "/control", {"action": "estop"})
|
|
assert resp["ok"] is False
|
|
finally:
|
|
srv.stop()
|