fgl-aircon/tests/ayla/test_session_mock.py

#!/usr/bin/env python3
"""Интеграционные сценарии сессии против mock-модуля (mock_ac.py).

usage: test_session_mock.py <session_runner> <mock_ac.py>
Сценарии: normal, rekey, fail_push, slot_503, no_poll, bad_ke, set_get.
"""
import os
import socket
import subprocess
import sys
import tempfile
import threading
import time

RUNNER = sys.argv[1]
MOCK = sys.argv[2]

# Синтетический ключ (НЕ боевой).
LANIP_KEY = "TW9ja0tleUFDbkdvMTIzNDU2Nzg5MDEyMw=="
KEY_ID = 64201
DSN = "AC000W00MOCK0001"


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


class Proc:
    """Процесс с построчным логом stdout."""

    def __init__(self, argv, env=None):
        self.lines = []
        self.lock = threading.Lock()
        self.ev = threading.Event()
        self.proc = subprocess.Popen(
            argv, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            text=True)
        threading.Thread(target=self._reader, daemon=True).start()

    def _reader(self):
        for line in self.proc.stdout:
            with self.lock:
                self.lines.append(line.rstrip("\n"))
                self.ev.set()

    def wait_line(self, prefix, timeout, exclude_prefix=None):
        """Ждёт строку, начинающуюся с prefix (или содержащую, если prefix
        начинается с '~')."""
        contains = prefix.startswith("~")
        needle = prefix[1:] if contains else prefix
        deadline = time.time() + timeout
        seen = 0
        while time.time() < deadline:
            with self.lock:
                for i, line in enumerate(self.lines):
                    if i < seen:
                        continue
                    hit = (needle in line) if contains else line.startswith(prefix)
                    if hit:
                        if exclude_prefix and line.startswith(exclude_prefix):
                            continue
                        return line
                    seen = i + 1
            self.ev.wait(0.2)
        return None

    def all_lines(self):
        with self.lock:
            return list(self.lines)

    def stop(self, timeout=10):
        self.proc.terminate()
        try:
            self.proc.wait(timeout)
        except subprocess.TimeoutExpired:
            self.proc.kill()


def run_mock(extra, port):
    p = Proc([sys.executable, MOCK, "--port", str(port), "--lanip-key", LANIP_KEY,
              "--key-id", str(KEY_ID)] + extra)
    if p.wait_line("READY", 10) is None:
        raise RuntimeError("mock не стартовал: " + "\n".join(p.all_lines()))
    return p


def run_runner(mock_port, listen_port, props, duration, env_extra=None,
               timeout=None):
    env = dict(os.environ)
    if env_extra:
        env.update(env_extra)
    argv = [RUNNER, "127.0.0.1", str(mock_port), str(listen_port), DSN,
            LANIP_KEY, str(KEY_ID), str(duration)] + props
    return Proc(argv, env=env), (timeout or duration + 25)


FAILURES = []


def check(cond, what):
    tag = "ok  " if cond else "FAIL"
    print(f"  [{tag}] {what}")
    if not cond:
        FAILURES.append(what)


def scenario_normal():
    print("== scenario: normal (установка, батч GET, push, delete)")
    mock_port, listen = free_port(), free_port()
    mock = run_mock([], mock_port)
    runner, timeout = run_runner(mock_port, listen,
                                 ["operation_mode", "fan_speed"], 3)
    line = runner.wait_line("STATE online", 10)
    check(line is not None, "переход в online")
    p1 = runner.wait_line("PROP operation_mode", 10)
    p2 = runner.wait_line("PROP fan_speed", 10)
    check(p1 is not None and "cid=" not in p1 and " i:6" in p1, f"push #1: {p1}")
    check(p2 is not None and " i:4" in p2, f"push #2: {p2}")
    check(runner.wait_line("DELETED", timeout) is not None, "delete_session")
    check(runner.wait_line("STATS", 5) is not None, "STATS")
    check("rekeys=1" in " ".join(runner.all_lines()), "ровно 1 key exchange")
    check("CMD GET operation_mode" in "\n".join(mock.all_lines()),
          "mock исполнил GET")
    check("DELETE" in "\n".join(mock.all_lines()), "mock получил DELETE")
    runner.stop()
    mock.stop()


def scenario_rekey():
    print("== scenario: rekey по local_reg (chain непрерывна)")
    mock_port, listen = free_port(), free_port()
    mock = run_mock(["--rekey-every", "2", "--push-every", "1"], mock_port)
    runner, timeout = run_runner(mock_port, listen, [], 6,
                                 env_extra={"RUNNER_KEEPALIVE_MS": "1000"})
    check(runner.wait_line("STATE online", 10) is not None, "online")
    t1 = runner.wait_line("PROP tick", 15)
    t2 = runner.wait_line("PROP tick", 15)
    check(t1 is not None and t2 is not None, "спонтанные push после re-key")
    runner.wait_line("DELETED", timeout)
    stats = "\n".join(runner.all_lines())
    check("rekeys=" in stats and "pushes_bad=0" in stats,
          f"без потерянных push: {stats.splitlines()[-1] if stats else '?'}")
    check("CMD bad-sign" not in "\n".join(mock.all_lines()),
          "mock проверял подписи наших команд")
    runner.stop()
    mock.stop()


def scenario_fail_push():
    print("== scenario: push с битой подписью -> recovering -> online")
    mock_port, listen = free_port(), free_port()
    mock = run_mock(["--push-every", "1", "--fail-pushes", "1"], mock_port)
    runner, timeout = run_runner(mock_port, listen, [], 8,
                                 env_extra={"RUNNER_KEEPALIVE_MS": "1000"})
    check(runner.wait_line("STATE online", 10) is not None, "online")
    rec = runner.wait_line("STATE recovering 6", 15)
    check(rec is not None, f"recovering после битой подписи: {rec}")
    # Следующий push расшифровывается (цепочка в синке) -> online.
    check(runner.wait_line("STATE online 0", 15) is not None,
          "online восстановлен следующим валидным push")
    check(runner.wait_line("STATS", timeout) is not None, "STATS получены")
    stats = "\n".join(runner.all_lines())
    stats_line = [l for l in stats.splitlines() if l.startswith("STATS")]
    check("pushes_bad=1" in stats, f"учтён 1 битый push: {stats_line}")
    runner.stop()
    mock.stop()


def scenario_slot_503():
    print("== scenario: 503 — нет слотов")
    mock_port, listen = free_port(), free_port()
    mock = run_mock(["--503"], mock_port)
    runner, timeout = run_runner(mock_port, listen, [], 4,
                                 env_extra={"RUNNER_NOSLOT_RETRY_MS": "2000"})
    line = runner.wait_line("STATE offline 1", 10)
    check(line is not None, "offline с ошибкой NoSlot")
    regs = [l for l in mock.all_lines() if l.startswith("REG")]
    check(len(regs) >= 1, "local_reg доходил до модуля")
    runner.stop()
    mock.stop()


def scenario_no_poll():
    print("== scenario: KE без poll — активация не наступила, тихая пауза")
    mock_port, listen = free_port(), free_port()
    mock = run_mock(["--no-poll"], mock_port)
    runner, timeout = run_runner(mock_port, listen, [], 5,
                                 env_extra={"RUNNER_ACTIVATION_MS": "1500",
                                            "RUNNER_QUIET_MS": "3000",
                                            "RUNNER_KEEPALIVE_MS": "800"})
    line = runner.wait_line("STATE recovering 5", 15)
    check(line is not None, "recovering с ActivationTimeout")
    time.sleep(1.0)
    regs = [l for l in mock.all_lines() if l.startswith("REG")]
    # Первый reg + возможно второй до таймаута активации; после — тишина.
    check(len(regs) <= 3, f"нет спама local_reg (получено {len(regs)})")
    runner.stop()
    mock.stop()


def scenario_bad_ke():
    print("== scenario: key_exchange ver=2 -> key_error")
    mock_port, listen = free_port(), free_port()
    mock = run_mock(["--fail-first-ke"], mock_port)
    runner, timeout = run_runner(mock_port, listen, [], 3)
    line = runner.wait_line("STATE key_error 4", 10)
    check(line is not None, "key_error с BadKeyExchange")
    check(mock.wait_line("~-> 426", 10) is not None, "mock получил 426")
    runner.stop()
    mock.stop()


def scenario_chain_divergence_heals():
    print("== scenario: входящая цепочка расходится на 1 сообщение и сходится")
    mock_port, listen = free_port(), free_port()
    mock = run_mock(["--push-every", "1", "--garbage-pushes", "1"], mock_port)
    runner, timeout = run_runner(mock_port, listen, [], 8,
                                 env_extra={"RUNNER_KEEPALIVE_MS": "1000"})
    check(runner.wait_line("STATE online", 10) is not None, "online")
    check(runner.wait_line("STATE recovering 6", 15) is not None,
          "recovering после урезанного push")
    # CBC-состояние = последний шифроблок: следующий полный push сходится.
    check(runner.wait_line("STATE online 0", 15) is not None,
          "online восстановлен следующим push (цепочка сошлась)")
    stats = runner.wait_line("STATS", timeout)
    # Урезанный push + один residual (первый блок следующего) = 2 потери,
    # далее цепочка сошлась (состояние CBC = последний шифроблок).
    check(stats is not None and "pushes_bad=2" in stats,
          f"ровно 2 потери: {stats}")
    runner.stop()
    mock.stop()


def scenario_outbound_desync_rekey():
    print("== scenario: исходящий десинк -> модуль ре-кает на local_reg")
    mock_port, listen = free_port(), free_port()
    mock = run_mock(["--break-outbound", "1"], mock_port)
    runner, timeout = run_runner(mock_port, listen, ["operation_mode"], 12,
                                 env_extra={"RUNNER_KEEPALIVE_MS": "1000",
                                            "RUNNER_SET_NAME": "fan_speed",
                                            "RUNNER_SET_VALUE": "2",
                                            "RUNNER_SET_AT": "6"})
    check(runner.wait_line("STATE online", 10) is not None, "online")
    check(mock.wait_line("~CMD skipped", 10) is not None,
          "мок пропустил ответ (десинк)")
    check(mock.wait_line("~CMD bad-sign", 15) is not None,
          "подпись следующей команды не сошлась")
    # Реальный модуль ре-кает на следующем local_reg (как в probe1-3);
    # после re-key цепочки свежие — SET доходит.
    check(mock.wait_line("~CMD SET fan_speed=2", 20) is not None,
          "SET доставлена после re-key")
    check(len([l for l in mock.all_lines() if l.startswith("KE ")]) >= 2,
          "второй key exchange был")
    runner.stop()
    mock.stop()


def scenario_set_get():
    print("== scenario: SET + подтверждение GET (эха нет)")
    mock_port, listen = free_port(), free_port()
    mock = run_mock([], mock_port)
    env = {"RUNNER_SET_NAME": "fan_speed", "RUNNER_SET_VALUE": "3",
           "RUNNER_SET_AT": "1"}
    runner, timeout = run_runner(mock_port, listen, [], 5, env_extra=env)
    line = runner.wait_line("SET_DONE fan_speed=3", 10)
    check(line is not None, "SET отправлена")
    check(mock.wait_line("CMD SET fan_speed=3", 10) is not None,
          "mock применил SET")
    runner.wait_line("DELETED", timeout)
    runner.stop()
    mock.stop()


def main():
    scenarios = [scenario_normal, scenario_rekey, scenario_fail_push,
                 scenario_slot_503, scenario_no_poll, scenario_bad_ke,
                 scenario_chain_divergence_heals,
                 scenario_outbound_desync_rekey, scenario_set_get]
    for sc in scenarios:
        sc()
    print()
    if FAILURES:
        print("ПРОВАЛЕНО:", len(FAILURES))
        for f in FAILURES:
            print(" -", f)
        sys.exit(1)
    print("ВСЕ СЦЕНАРИИ ПРОЙДЕНЫ")


if __name__ == "__main__":
    main()