dmr-server/tests/test_panel.py

"""Проверка панели: страницы, API, SSE и Last Heard.

Поднимает настоящий сервер и говорит с ним по HTTP, как браузер.
Запуск:  python3 tests/test_panel.py
"""
from __future__ import annotations

import json
import os
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
from base64 import b64encode
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

CFG = "/tmp/dmrsrv-panel-test.json"
PORT = 18080
UDP = 18031
ok_n = fail_n = 0


def ok(m):
    global ok_n
    ok_n += 1
    print(f"  \033[32mOK\033[0m   {m}")


def bad(m):
    global fail_n
    fail_n += 1
    print(f"  \033[31mFAIL\033[0m {m}")


def check(c, g, b):
    ok(g) if c else bad(b)


Path(CFG).write_text(json.dumps({
    "bind": "127.0.0.1", "port": UDP, "passphrase": "testpass",
    "panel": {"bind": "127.0.0.1", "port": PORT, "user": "admin", "password": "sekret"},
    "groups": [{"tg": 9, "name": "Локальная", "slot": 2}],
    "parrot": {"enabled": False, "tg": 9990},
    "hotspots": [], "allow_unknown_hotspots": True,
    "subscribers": {"mode": "off", "list": []},
}, ensure_ascii=False))

env = dict(os.environ, DMRSRV_CONFIG=CFG)
srv = subprocess.Popen([sys.executable, "-m", "dmrsrv"], cwd=ROOT, env=env,
                       stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
AUTH = "Basic " + b64encode(b"admin:sekret").decode()


def req(path, data=None, auth=True, timeout=6):
    r = urllib.request.Request(f"http://127.0.0.1:{PORT}{path}")
    if auth:
        r.add_header("Authorization", AUTH)
    if data is not None:
        r.data = json.dumps(data).encode()
        r.add_header("Content-Type", "application/json")
    return urllib.request.urlopen(r, timeout=timeout)


try:
    for _ in range(60):
        try:
            req("/api/state").read()
            break
        except Exception:
            time.sleep(0.25)
    else:
        print("сервер не поднялся:", srv.stdout.read() if srv.stdout else "")
        sys.exit(1)

    print("==> авторизация")
    try:
        req("/", auth=False)
        bad("панель отдаётся без пароля!")
    except urllib.error.HTTPError as e:
        check(e.code == 401, "без пароля — 401", f"код {e.code}")
    check(req("/").status == 200, "с паролем — 200", "с паролем не пускает")

    print("==> страницы")
    html = req("/").read().decode()
    for pid, name in [("p-overview", "Обзор"), ("p-lastheard", "Last Heard"),
                      ("p-hotspots", "Хотспоты"), ("p-groups", "Группы"),
                      ("p-access", "Доступ"), ("p-settings", "Настройки")]:
        check(f'id={pid}' in html, f"страница «{name}» есть", f"нет страницы {pid}")
    check("EventSource" in html, "журнал подключён живым потоком (SSE)", "SSE не подключён")

    print("==> управление хотспотами")
    req("/api/hotspots/add", {"peer_id": 2501001, "name": "Гараж", "password": "hotpass"})
    st = json.load(req("/api/state"))
    check(any(h["peer_id"] == 2501001 for h in st["hotspots"]), "хотспот добавлен", "не добавился")
    try:
        req("/api/hotspots/add", {"peer_id": 2501001})
        bad("дубль хотспота принят")
    except urllib.error.HTTPError as e:
        check(e.code == 400, "повторный номер отклонён", f"код {e.code}")
    req("/api/hotspots/toggle", {"peer_id": 2501001})
    st = json.load(req("/api/state"))
    check(st["hotspots"][0]["enabled"] is False, "хотспот отключается", "не отключился")

    print("==> списки абонентов")
    req("/api/subscribers", {"mode": "deny", "list": "2501111 2502000-2502005"})
    st = json.load(req("/api/state"))
    check(st["subscribers"]["mode"] == "deny", "режим сохранён", "режим не сохранён")
    check(len(st["subscribers"]["list"]) == 2, "список сохранён", "список не сохранён")
    try:
        req("/api/subscribers", {"mode": "allow", "list": "не-число"})
        bad("мусор в списке принят — оператор не увидит опечатку")
    except urllib.error.HTTPError as e:
        check(e.code == 400, "мусор в списке отклонён с объяснением", f"код {e.code}")
    try:
        req("/api/subscribers", {"mode": "странно", "list": ""})
        bad("неизвестный режим принят")
    except urllib.error.HTTPError as e:
        check(e.code == 400, "неизвестный режим отклонён", f"код {e.code}")

    print("==> Last Heard наполняется из эфира")
    # шлём настоящее рукопожатие и голосовой вызов
    import hashlib
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.settimeout(3)
    PEER = 2509999

    def rt(p):
        s.sendto(p, ("127.0.0.1", UDP))
        return s.recv(2048)

    salt = rt(b"RPTL" + PEER.to_bytes(4, "big"))[6:10]
    rt(b"RPTK" + PEER.to_bytes(4, "big") + hashlib.sha256(salt + b"testpass").digest())
    body = (f"{'LHTEST':<8.8s}{438000000:09d}{430000000:09d}{5:02d}{1:02d}"
            f"{'51.66600':8.8s}{'-1.170000':9.9s}{10:03d}{'Irkutsk':<20.20s}"
            f"{'Russia':<19.19s}4{'https://x':<124.124s}{'MMDVM_v1':<40.40s}{'MMDVM_HS':<40.40s}")
    rt(b"RPTC" + PEER.to_bytes(4, "big") + body.encode())

    def frame(seq, dtype, ftype=0b10):
        bits = 0x80 | (ftype << 4) | dtype
        return (b"DMRD" + bytes([seq]) + (2501234).to_bytes(3, "big") + (9).to_bytes(3, "big")
                + PEER.to_bytes(4, "big") + bytes([bits]) + b"\xaa\xbb\xcc\xdd"
                + bytes(range(33)) + b"\x00\x00")

    s.sendto(frame(0, 0x01), ("127.0.0.1", UDP))          # начало
    for i in range(1, 6):
        s.sendto(frame(i, i % 6, 0b00), ("127.0.0.1", UDP))
    s.sendto(frame(6, 0x02), ("127.0.0.1", UDP))          # терминатор
    time.sleep(1.5)

    st = json.load(req("/api/state"))
    calls = st.get("calls", [])
    check(len(calls) >= 1, "вызов попал в Last Heard", "Last Heard пуст после реального вызова")
    if calls:
        c = calls[0]
        check(c["src_id"] == 2501234, "источник записан верно", f"src={c['src_id']}")
        check(c["dst_id"] == 9, "назначение записано верно", f"dst={c['dst_id']}")
        check(c["peer_id"] == PEER, "видно, через какой хотспот", f"peer={c['peer_id']}")
        check(not c["active"], "вызов закрыт по терминатору", "вызов остался активным")
        check(c["frames"] >= 6, "кадры посчитаны", f"кадров {c['frames']}")

    print("==> поля RPTC доходят до панели")
    p = st["peers"][0] if st["peers"] else {}
    check(p.get("country") == "Russia", "страна (Pi-Star кладёт её в description)",
          f"country={p.get('country')!r}")
    check(p.get("location") == "Irkutsk", "местоположение", f"location={p.get('location')!r}")
    check(p.get("color_code") == 1, "color code (раньше читался как colorcode и был пуст)",
          f"color_code={p.get('color_code')!r}")
    check(p.get("tx_power") == 5, "мощность", f"tx_power={p.get('tx_power')!r}")
    check(p.get("rx_freq") == 438000000, "частота RX", f"rx={p.get('rx_freq')!r}")
    check(p.get("lat") == 51.666, "широта", f"lat={p.get('lat')!r}")
    check(p.get("package_id") == "MMDVM_HS", "модель устройства", f"pkg={p.get('package_id')!r}")

    print("==> живой поток событий (SSE)")
    r = urllib.request.Request(f"http://127.0.0.1:{PORT}/api/events")
    r.add_header("Authorization", AUTH)
    resp = urllib.request.urlopen(r, timeout=6)
    check(resp.headers.get("Content-Type", "").startswith("text/event-stream"),
          "заголовок text/event-stream", f"тип {resp.headers.get('Content-Type')}")
    line = resp.readline().decode()
    check(line.startswith("data: "), "события приходят в формате SSE", f"строка {line!r}")
    resp.close()

finally:
    srv.terminate()
    try:
        srv.wait(timeout=5)
    except subprocess.TimeoutExpired:
        srv.kill()

print(f"\n  \033[1m{ok_n} OK, {fail_n} FAIL\033[0m\n")
sys.exit(0 if fail_n == 0 else 1)