fgl-aircon/tests/tools/test_fglair_discover.py

#!/usr/bin/env python3
"""Тест fglair-discover против мок-облака (stdlib http.server).

Мок эмулирует: /users/sign_in.json, /apiv1/devices.json,
/apiv1/dsns/<dsn>/lan.json (PROTOCOL.md §7). Проверяет оба формата вывода,
фильтр --device, отказ с 401.
"""
import json
import subprocess
import sys
import threading
import unittest
import http.server
import urllib.parse

import os
DISCOVER = os.path.join(os.path.dirname(__file__), "..", "..",
                        "tools", "fglair-discover")

TOKEN = "test-token-123"
DEVICES = [
    {"device": {"dsn": "AC000W00TEST0001", "product_name": "Living Room",
                "oem_model": "AP-WC1E", "mac": "A0:C9:A0:0D:61:C9",
                "lan_ip": "192.0.2.10"}},
    {"device": {"dsn": "AC000W00TEST0002", "product_name": "Bedroom",
                "oem_model": "AP-WB2E", "mac": "A0:C9:A0:0D:61:CA",
                "lan_ip": "192.0.2.11"}},
]
LANIPS = {
    "AC000W00TEST0001": {"lanip": {"lanip_key": "KEY1abc==", "lanip_key_id": 111}},
    "AC000W00TEST0002": {"lanip": {"lanip_key": "KEY2def==", "lanip_key_id": 222}},
}


class MockCloud(http.server.BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def _json(self, code, obj):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        n = int(self.headers.get("Content-Length") or 0)
        body = json.loads(self.rfile.read(n) or b"{}")
        if self.path == "/users/sign_in.json":
            app = body.get("user", {}).get("application", {})
            if app.get("app_id") == "FGLair-eu-id" and \
               app.get("app_secret") and \
               body["user"]["password"] == "pw":
                self._json(200, {"access_token": TOKEN})
            else:
                self._json(401, {"error": "invalid_credentials"})
        else:
            self._json(404, {})

    def do_GET(self):
        if not self.headers.get("Authorization") == f"auth_token {TOKEN}":
            self._json(401, {"error": "auth_required"})
            return
        if self.path == "/apiv1/devices.json":
            self._json(200, DEVICES)
            return
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path.startswith("/apiv1/dsns/") and \
           parsed.path.endswith("/lan.json"):
            dsn = parsed.path.split("/")[3]
            self._json(200, LANIPS.get(dsn, {"lanip": {}}))
            return
        self._json(404, {})


def run_discover(extra):
    return subprocess.run(
        [sys.executable, DISCOVER, "--region", "eu",
         "--email", "u@x.y", "--password", "pw"] + extra,
        capture_output=True, text=True, timeout=30)


class TestDiscover(unittest.TestCase):

    @classmethod
    def setUpClass(cls):
        cls.srv = http.server.ThreadingHTTPServer(("127.0.0.1", 0), MockCloud)
        cls.port = cls.srv.server_address[1]
        threading.Thread(target=cls.srv.serve_forever, daemon=True).start()

    @classmethod
    def tearDownClass(cls):
        cls.srv.shutdown()

    def _base(self, extra=None):
        return ["--insecure", "--api-base", f"http://127.0.0.1:{self.port}"] \
            + (extra or [])

    def test_json_full(self):
        r = run_discover(self._base())
        self.assertEqual(r.returncode, 0, r.stderr)
        cfgs = [json.loads(l) for l in r.stdout.strip().splitlines()]
        self.assertEqual(len(cfgs), 2)
        c = cfgs[0]
        self.assertEqual(c["dsn"], "AC000W00TEST0001")
        self.assertEqual(c["lanip_key"], "KEY1abc==")
        self.assertEqual(c["lanip_key_id"], 111)
        self.assertEqual(c["model"], "AP-WC1E")
        self.assertEqual(c["temp_type"], "C")
        self.assertEqual(c["mac_address"], "a0c9a00d61c9")

    def test_device_filter(self):
        r = run_discover(self._base(["--device", "Bedroom"]))
        self.assertEqual(r.returncode, 0, r.stderr)
        cfgs = [json.loads(l) for l in r.stdout.strip().splitlines()]
        self.assertEqual(len(cfgs), 1)
        self.assertEqual(cfgs[0]["dsn"], "AC000W00TEST0002")

    def test_esphome_secrets(self):
        r = run_discover(self._base(["--device", "Living Room",
                                     "--format", "esphome-secrets"]))
        self.assertEqual(r.returncode, 0, r.stderr)
        self.assertIn('livingroom_dsn: "AC000W00TEST0001"', r.stdout)
        self.assertIn('livingroom_lanip_key: "KEY1abc=="', r.stdout)
        self.assertIn("livingroom_lanip_key_id: 111", r.stdout)

    def test_bad_password(self):
        r = subprocess.run(
            [sys.executable, DISCOVER, "--region", "eu", "--email", "u@x.y",
             "--password", "wrong", "--insecure",
             "--api-base", f"http://127.0.0.1:{self.port}"],
            capture_output=True, text=True, timeout=30)
        self.assertNotEqual(r.returncode, 0)
        self.assertIn("Ошибка входа", r.stderr)


if __name__ == "__main__":
    unittest.main()