fgl-aircon/tests/acceptance/test_acceptance.py

"""Self-test приёмочного скрипта на моке HA REST API (без железа)."""
from __future__ import annotations

import json
import pathlib
import sys
import threading
import types
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from types import SimpleNamespace

import pytest

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

from test_esphome_ha import (  # noqa: E402
    AcceptanceError,
    EsphomeChannel,
    HaRest,
    Step,
    build_steps,
    main,
)


class MockHa:
    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.fail_at = 0
        self.post_count = 0
        self.ignore_services = False
        self.non_json = False
        self.state = {
            "entity_id": "climate.test",
            "state": "heat",
            "attributes": {
                "hvac_modes": ["off", "cool", "heat"],
                "fan_modes": ["low", "high"],
                "swing_modes": ["off", "on"],
                "temperature": 22.0,
                "min_temp": 16.0,
                "max_temp": 30.0,
                "fan_mode": "low",
                "swing_mode": "off",
            },
        }

    def apply(self, service: str, data: dict) -> None:
        attrs = self.state["attributes"]
        if service == "set_hvac_mode":
            self.state["state"] = data["hvac_mode"]
        elif service == "set_fan_mode":
            attrs["fan_mode"] = data["fan_mode"]
        elif service == "set_swing_mode":
            attrs["swing_mode"] = data["swing_mode"]
        elif service == "set_temperature":
            attrs["temperature"] = data["temperature"]

    def handler(self):
        mock = self

        class Handler(BaseHTTPRequestHandler):
            protocol_version = "HTTP/1.1"

            def log_message(self, *args):
                pass

            def _json(self, status: int, payload) -> None:
                body = json.dumps(payload).encode()
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Connection", "close")
                self.end_headers()
                self.close_connection = True
                self.wfile.write(body)

            def _raw(self, status: int, body: bytes) -> None:
                self.send_response(status)
                self.send_header("Content-Type", "text/plain")
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Connection", "close")
                self.end_headers()
                self.close_connection = True
                self.wfile.write(body)

            def do_GET(self):
                if self.path != "/api/states/climate.test":
                    self._json(404, {"message": "not found"})
                    return
                with mock.lock:
                    if mock.non_json:
                        self._raw(200, b"<html>captive portal</html>")
                        return
                    self._json(200, json.loads(json.dumps(mock.state)))

            def do_POST(self):
                length = int(self.headers.get("Content-Length") or 0)
                data = json.loads(self.rfile.read(length) or b"{}")
                with mock.lock:
                    mock.post_count += 1
                    if mock.fail_at and mock.post_count == mock.fail_at:
                        self._json(500, {"message": "boom"})
                        return
                    if not mock.ignore_services:
                        mock.apply(self.path.rsplit("/", 1)[-1], data)
                    self._json(200, [])

        return Handler


@pytest.fixture
def mock_ha():
    mock = MockHa()
    server = ThreadingHTTPServer(("127.0.0.1", 0), mock.handler())
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    port = server.server_address[1]
    yield mock, f"http://127.0.0.1:{port}"
    server.shutdown()
    server.server_close()
    thread.join(timeout=5)


QUICK_ARGS = ["quick", "--ha-token", "t", "--entity", "climate.test"]


def test_build_steps_supported_only():
    state = {
        "state": "heat",
        "attributes": {
            "hvac_modes": ["off"],
            "fan_modes": [],
            "temperature": 20.0,
            "min_temp": 16.0,
            "max_temp": 30.0,
        },
    }
    assert [step.name for step in build_steps(state)] == ["temperature"]


def test_quick_success(mock_ha, capsys, tmp_path):
    _, url = mock_ha
    report = tmp_path / "quick.csv"
    assert main(
        [*QUICK_ARGS, "--ha-url", url, "--report", str(report)]
    ) == 0
    out = capsys.readouterr().out
    assert "restore: expected=heat observed=heat" in out
    assert "ok" in report.read_text()


def test_quick_burst_success(mock_ha):
    _, url = mock_ha
    assert main([*QUICK_ARGS, "--burst", "--ha-url", url]) == 0


def test_quick_service_error(mock_ha, capsys):
    mock, url = mock_ha
    mock.fail_at = 1
    assert main([*QUICK_ARGS, "--ha-url", url]) == 2
    assert "HTTP 500" in capsys.readouterr().err


def test_quick_restores_after_partial_failure(mock_ha):
    """Сбой на втором шаге — исходное состояние возвращается (finally)."""
    mock, url = mock_ha
    mock.fail_at = 2
    assert main([*QUICK_ARGS, "--ha-url", url]) == 2
    assert mock.state["state"] == "heat"
    assert mock.state["attributes"]["fan_mode"] == "low"


def test_quick_timeout_is_failure(mock_ha, capsys):
    mock, url = mock_ha
    mock.ignore_services = True
    rc = main(
        [*QUICK_ARGS, "--ha-url", url, "--settle-timeout", "0.3"]
    )
    assert rc == 1
    assert "ПРОВАЛЕНО" in capsys.readouterr().err


def test_quick_non_json_answer(mock_ha, capsys):
    mock, url = mock_ha
    mock.non_json = True
    assert main([*QUICK_ARGS, "--ha-url", url]) == 2
    assert "не JSON" in capsys.readouterr().err


def test_long_report_on_success(mock_ha, tmp_path, capsys):
    _, url = mock_ha
    report = tmp_path / "acceptance.csv"
    rc = main([
        "long", "--ha-url", url, "--ha-token", "t",
        "--entity", "climate.test",
        "--hours", "0.000003", "--interval", "0.01",
        "--report", str(report),
    ])
    assert rc == 0
    text = report.read_text()
    assert "timestamp,step,expected,observed,result" in text
    assert "ok" in text
    assert "long: ok=1 fail=0" in capsys.readouterr().out


def test_long_failure_restores_and_reports(mock_ha, tmp_path):
    mock, url = mock_ha
    mock.ignore_services = True  # шаг не достигается, восстановление — работает
    report = tmp_path / "fail.csv"
    rc = main([
        "long", "--ha-url", url, "--ha-token", "t",
        "--entity", "climate.test",
        "--hours", "0.000003", "--interval", "0.01",
        "--settle-timeout", "0.3",
        "--report", str(report),
    ])
    assert rc == 1
    assert "fail" in report.read_text()
    assert mock.state["state"] == "heat"


def test_wait_state_timeout(mock_ha):
    _, url = mock_ha
    ha = HaRest(url, "t", timeout=1)
    with pytest.raises(AcceptanceError):
        ha.wait_state("climate.test", lambda s: False, timeout=0.2)


def test_esphome_channel_sends_mode_and_temperature(monkeypatch):
    instances: list = []

    class ClimateInfo:
        key = 7

    class FakeClient:
        def __init__(self, host, port, noise_psk=None):
            self.calls: list = []
            instances.append(self)

        async def connect(self, login=False):
            self.login = login

        async def list_entities_services(self):
            return [ClimateInfo()], []

        async def climate_command(self, key=None, mode=None,
                                  target_temperature=None):
            self.calls.append((key, mode, target_temperature))

        async def disconnect(self):
            self.disconnected = True

    fake = types.ModuleType("aioesphomeapi")
    fake.APIClient = FakeClient
    fake.ClimateMode = SimpleNamespace(COOL="COOL", HEAT="HEAT")
    fake.ClimateInfo = ClimateInfo
    monkeypatch.setitem(sys.modules, "aioesphomeapi", fake)

    channel = EsphomeChannel("esphome.local", "key")
    try:
        channel.connect()
        mode_step = Step("hvac_mode", "climate", "set_hvac_mode",
                         {"hvac_mode": "cool"}, "state", "cool")
        temp_step = Step("temperature", "climate", "set_temperature",
                         {"temperature": 23.5}, "temperature", 23.5)
        fan_step = Step("fan_mode", "climate", "set_fan_mode",
                        {"fan_mode": "low"}, "fan_mode", "low")
        assert channel.apply(mode_step) is True
        assert channel.apply(temp_step) is True
        assert channel.apply(fan_step) is False
    finally:
        channel.close()

    assert instances[0].calls == [(7, "COOL", None), (7, None, 23.5)]
    assert instances[0].login is True
    assert instances[0].disconnected is True