ros_control/tests/test_devices.py

import json
import re

import httpx
import pytest
from fastapi.testclient import TestClient

from app.db import session_scope
from app.main import create_app
from app.models import Device
from app.ros import operations as ros
from app.services import devices, groups, ops
from tests.conftest import ADMIN_PASSWORD, API_TOKEN
from tests.helpers import ros_client


@pytest.mark.asyncio
async def test_status_parsing():
    data = {
        "/rest/system/resource": {"board-name": "hAP ax3", "version": "7.15 (stable)", "uptime": "1d2h"},
        "/rest/system/routerboard": {"routerboard": "true", "model": "C53UiG", "current-firmware": "7.14", "upgrade-firmware": "7.15"},
        "/rest/system/package/update": {"channel": "stable", "installed-version": "7.15", "latest-version": "7.16", "status": "New version is available"},
        "/rest/system/package/update/check-for-updates": [],  # POST: проверка обновлений
    }
    async with ros_client(lambda r: httpx.Response(200, json=data[r.url.path])) as c:
        st = await ros.get_status(c)
    assert (st["model"], st["channel"], st["fw_upgrade"], st["ros_latest"]) == ("C53UiG", "stable", "7.15", "7.16")
    assert st["ros_installed"] == "7.15" and st["ros_check_error"] is None


@pytest.mark.asyncio
async def test_chr_status_and_version_compare():
    """CHR: раздела system/routerboard нет (HTTP 400) — это не сбой; «последняя» версия старше установленной — не обновление."""
    def handler(r: httpx.Request) -> httpx.Response:
        if r.url.path == "/rest/system/routerboard":
            return httpx.Response(400, json={"detail": "no such command or directory (routerboard)", "error": 400})
        data = {
            "/rest/system/resource": {"board-name": "CHR Yandex epyc-9654", "version": "7.24 (stable)", "uptime": "1d"},
            "/rest/system/package/update": {"channel": "long-term", "installed-version": "7.24", "latest-version": "7.23.7"},
        }
        return httpx.Response(200, json=data.get(r.url.path, []))

    async with ros_client(handler) as c:
        st = await ros.get_status(c)
        assert await ros.upgrade_firmware(c) == "Устройство без RouterBOARD firmware (например, CHR) — пропущено"
    assert st["model"].startswith("CHR") and st["fw_current"] is None
    assert not ros.version_newer(st["ros_latest"], st["ros_installed"])  # 7.23.7 старше 7.24
    assert ros.version_newer("7.24.1", "7.24") and ros.version_newer("7.25", "7.25rc1") and not ros.version_newer("7.25rc1", "7.25")


def test_groups_and_device_filters():
    office, store = groups.create_group("Офис"), groups.create_group("Склад")
    a = devices.create_device("a", "10.0.0.1", 80, "u", "p", group_id=office.id)
    b = devices.create_device("b", "10.0.0.2", 80, "u", "p", group_id=store.id)
    devices.create_device("c", "10.0.0.3", 80, "u", "p")
    with session_scope() as s:
        s.get(Device, a.id).online = True
        s.get(Device, a.id).status_json = json.dumps({"ros_installed": "7.1", "ros_latest": "7.2", "channel": "stable"})
        s.get(Device, b.id).online = False

    def names(**kw):
        return sorted(d.name for d in devices.filter_devices(devices.list_devices(), **kw))

    assert names(group=str(office.id)) == ["a"] and names(group="none") == ["c"]
    assert names(status="offline") == ["b", "c"] and names(updates="ros") == ["a"]
    assert names(q="10.0.0.2") == ["b"] and names(channel="stable") == ["a"]

    groups.delete_group(office.id)  # устройства остаются, но без группы
    assert devices.get_device(a.id).group_id is None


def test_create_device_via_ui_binds_group(monkeypatch):
    """Добавление устройства из окна: привязка к существующей группе и создание группы «на месте»."""
    async def no_poll(_device_id):  # без обращения к реальному устройству
        return None

    monkeypatch.setattr(ops, "refresh_status", no_poll)
    office = groups.create_group("Офис")
    form = dict(host="10.0.0.1", port="80", username="u", password="p")
    htmx = {"HX-Request": "true"}
    with TestClient(create_app()) as c:
        c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD})
        # 1) существующая группа
        r = c.post("/devices/new", data={**form, "name": "a", "group_id": str(office.id), "note": "  Серверная, 2 этаж  "},
                   follow_redirects=False)
        assert r.status_code == 303
        # имя устройства в таблице открывает окно изменения (без JS — запасную страницу)
        page = c.get("/").text
        assert re.search(r'class="dev-link" href="/devices/(dev_[0-9a-f-]+)/edit" hx-get="/ui/dialog/device/\1"', page)
        # 2) «Новая группа…» из окна: группа создаётся и привязывается, окно просит обновить страницу
        r = c.post("/devices/new", data={**form, "name": "b", "group_id": "__new__", "new_group": "Склад"}, headers=htmx)
        assert r.headers["HX-Refresh"] == "true"
        # 3) «Новая группа…» без названия: ошибка в окне, устройство не создаётся
        r = c.post("/devices/new", data={**form, "name": "c", "group_id": "__new__", "new_group": ""}, headers=htmx)
        assert "Введите название новой группы" in r.text and 'id="modal-title"' in r.text

        # окна групп открываются и показывают ошибку (дубликат) внутри окна
        assert "Новая группа" in c.get("/ui/dialog/group").text
        r = c.post("/groups/new", data={"name": "Склад"}, headers=htmx)
        assert "уже существует" in r.text and 'id="modal-title"' in r.text

    by_name = {d.name: d for d in devices.list_devices()}
    store = next(g for g in groups.list_groups() if g["name"] == "Склад")
    assert by_name["a"].group_id == office.id and by_name["b"].group_id == store["id"] and "c" not in by_name
    assert by_name["a"].note == "Серверная, 2 этаж" and by_name["b"].note is None  # пробелы обрезаются, пусто -> None


def test_device_name_is_immutable():
    """Имя задаётся только при создании: API отклоняет смену, форма изменения имя игнорирует."""
    with TestClient(create_app()) as c:
        c.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD})
        h = {"Authorization": f"Bearer {API_TOKEN}"}
        d = c.post("/api/v1/devices", headers=h, json={"name": "r1", "host": "10.0.0.1", "username": "u", "password": "p"}).json()
        r = c.patch(f"/api/v1/devices/{d['id']}", headers=h, json={"name": "r2"})
        assert r.status_code == 400 and "нельзя изменить" in r.text
        # то же имя — допустимо (клиенты часто присылают объект целиком); остальные поля меняются
        r = c.patch(f"/api/v1/devices/{d['id']}", headers=h, json={"name": "r1", "host": "10.0.0.9"})
        assert r.status_code == 200 and r.json()["name"] == "r1" and r.json()["host"] == "10.0.0.9"
        # форма изменения: присланное имя игнорируется, остальное сохраняется
        r = c.post(f"/devices/{d['id']}/edit", data={"name": "hacked", "host": "10.0.0.7", "port": "80", "username": "u"},
                   follow_redirects=False)
        assert r.status_code == 303
        page = c.get(f"/ui/dialog/device/{d['id']}").text
        assert "readonly" in page and 'name="name"' not in page
    dev = devices.get_device(d["id"])
    assert (dev.name, dev.host) == ("r1", "10.0.0.7")


@pytest.mark.asyncio
async def test_poll_marks_offline_and_recovers(monkeypatch):
    """Фоновый опрос: недоступность видна сразу; после возвращения — полный опрос (версии обновились)."""
    d = devices.create_device("r1", "10.0.0.1", 80, "u", "p")
    up = {"on": False}
    data = {
        "/rest/system/resource": {"version": "7.24.4 (stable)", "uptime": "1m", "board-name": "hAP"},
        "/rest/system/routerboard": {"routerboard": "false"},
        "/rest/system/package/update": {"channel": "stable", "installed-version": "7.24.4", "latest-version": "7.24.4"},
    }

    def handler(r: httpx.Request) -> httpx.Response:
        if not up["on"]:
            raise httpx.ConnectError("connection refused")
        return httpx.Response(200, json=data.get(r.url.path, []))

    monkeypatch.setattr(devices, "open_client", lambda conn: ros_client(handler))
    await ops.poll_device(d.id)
    assert devices.get_device(d.id).online is False

    up["on"] = True
    await ops.poll_device(d.id)  # был offline -> полный опрос
    dev = devices.get_device(d.id)
    assert dev.online is True and dev.status["ros_installed"] == "7.24.4"

    data["/rest/system/resource"]["uptime"] = "2m"
    await ops.poll_device(d.id)  # онлайн -> лёгкий опрос обновляет uptime, остальное сохраняется
    dev = devices.get_device(d.id)
    assert dev.status["uptime"] == "2m" and dev.status["ros_latest"] == "7.24.4"


def test_ros_state_values():
    """Состояния колонки «Upgrade ROS»: unknown / check_error / update / downgrade / current."""
    assert devices.ros_state({}) == "unknown"
    assert devices.ros_state({"ros_installed": "7.24.4"}) == "unknown"  # нет latest — проверка ещё не выполнялась
    assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.16", "ros_check_error": "timeout"}) == "check_error"
    assert devices.ros_state({"ros_installed": "7.15", "ros_latest": "7.16"}) == "update"
    assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.23.7"}) == "downgrade"  # long-term старше установленной
    assert devices.ros_state({"ros_installed": "7.24.4", "ros_latest": "7.24.4"}) == "current"


def test_fw_state_values():
    """Состояния колонки «Upgrade FW» (изменение 027): unknown / update / downgrade / current;
    has_fw_update ложно для downgrade — откат прошивки не считается обновлением."""
    assert devices.fw_state({}) == "unknown"  # CHR или ещё не опрошено
    assert devices.fw_state({"fw_current": "7.14"}) == "unknown"  # нет upgrade-firmware
    assert devices.fw_state({"fw_current": "7.14", "fw_upgrade": "7.15"}) == "update"
    assert devices.fw_state({"fw_current": "7.24.4", "fw_upgrade": "7.23.7"}) == "downgrade"  # после отката ROS
    assert devices.fw_state({"fw_current": "7.15", "fw_upgrade": "7.15"}) == "current"
    assert devices.has_fw_update({"fw_current": "7.14", "fw_upgrade": "7.15"}) is True
    assert devices.has_fw_update({"fw_current": "7.24.4", "fw_upgrade": "7.23.7"}) is False