fgl-aircon/tests/acceptance/test_acceptance.py

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

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

import pytest

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

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


def free_port() -> int:
    sock = socket.socket()
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()
    return port


class MockHa:
    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.fail_next = 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 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 do_GET(self):
                if self.path == "/api/states/climate.test":
                    with mock.lock:
                        self._json(200, json.loads(json.dumps(mock.state)))
                else:
                    self._json(404, {"message": "not found"})

            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:
                    if mock.fail_next:
                        mock.fail_next = False
                        self._json(500, {"message": "boom"})
                        return
                    service = self.path.rsplit("/", 1)[-1]
                    attrs = mock.state["attributes"]
                    if service == "set_hvac_mode":
                        mock.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"]
                    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)


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,
        },
    }
    steps = build_steps(state)
    assert [step.name for step in steps] == ["temperature"]


def test_quick_success(mock_ha, capsys):
    _, url = mock_ha
    assert main(["quick", "--ha-url", url, "--ha-token", "t",
                 "--entity", "climate.test"]) == 0
    out = capsys.readouterr().out
    assert "restore: expected=heat observed=heat" in out


def test_quick_burst_success(mock_ha):
    _, url = mock_ha
    assert main(["quick", "--burst", "--ha-url", url, "--ha-token", "t",
                 "--entity", "climate.test"]) == 0


def test_quick_service_error(mock_ha, capsys):
    mock, url = mock_ha
    mock.fail_next = True
    assert main(["quick", "--ha-url", url, "--ha-token", "t",
                 "--entity", "climate.test"]) == 2
    assert "HTTP 500" in capsys.readouterr().err


def test_long_report(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_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)