petr.polezhaev
b84bd8d555
- C1: единственный владелец stop+destroy — reader-поток; close() только выставляет флаг и ждёт; повторный close и close из колбэка безопасны (регресс-тесты; TSan на close с мёртвым модулем — 0 гонок) - M1: wheel собирается с bundled mbedtls (нет внешней soname mbedcrypto); попутно исправлена bundled-ветка CMake (-lSTATIC) - C-API: счётчики потерянных событий/логов, лог-кольцо без union-балласта, атомарные fn/ctx лог-синка - provision: валидация формата облачных ответов; CLI: человеческие ошибки (exit 2), --out с правами 0600, console_script pyfglair - упаковка: MANIFEST.in (sdist с C++-исходниками), очистка stale .so, readme в метаданных; find_library без ложных .bak - тесты: close-сценарии, доставка логов, ошибки CLI/provision, C-API очередь
96 lines
2.7 KiB
Python
96 lines
2.7 KiB
Python
"""CLI pyfglair: discover против мок-облака и monitor против mock_ac."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
from pyfglair.__main__ import main
|
|
|
|
from conftest import free_port
|
|
|
|
|
|
async def test_cli_discover_json(cloud, capfd):
|
|
rc = await asyncio.to_thread(
|
|
main,
|
|
[
|
|
"discover", "--api-base", cloud["url"],
|
|
"--email", "user@example.com", "--password", "secret",
|
|
],
|
|
)
|
|
assert rc == 0
|
|
out, _ = capfd.readouterr()
|
|
config = json.loads(out.splitlines()[0])
|
|
assert config["dsn"] == "AC000W00MOCK0001"
|
|
assert config["lanip_key_id"] == 64201
|
|
assert config["ip_address"] == "192.168.1.50"
|
|
|
|
|
|
async def test_cli_discover_esphome_secrets(cloud, capfd):
|
|
rc = await asyncio.to_thread(
|
|
main,
|
|
[
|
|
"discover", "--api-base", cloud["url"],
|
|
"--email", "user@example.com", "--password", "secret",
|
|
"--format", "esphome-secrets",
|
|
],
|
|
)
|
|
assert rc == 0
|
|
out, _ = capfd.readouterr()
|
|
assert 'bedroom_dsn: "AC000W00MOCK0001"' in out
|
|
assert "bedroom_lanip_key_id: 64201" in out
|
|
|
|
|
|
async def test_cli_discover_error(cloud, capfd):
|
|
cloud["fail_login"] = True
|
|
rc = await asyncio.to_thread(
|
|
main,
|
|
[
|
|
"discover", "--api-base", cloud["url"],
|
|
"--email", "user@example.com", "--password", "bad",
|
|
],
|
|
)
|
|
assert rc == 2
|
|
err = capfd.readouterr().err
|
|
assert "Ошибка входа" in err
|
|
assert "Traceback" not in err
|
|
|
|
|
|
async def test_cli_discover_out_file(cloud, capfd, tmp_path):
|
|
out_file = tmp_path / "config_bedroom.json"
|
|
rc = await asyncio.to_thread(
|
|
main,
|
|
[
|
|
"discover", "--api-base", cloud["url"],
|
|
"--email", "user@example.com", "--password", "secret",
|
|
"--out", str(out_file),
|
|
],
|
|
)
|
|
assert rc == 0
|
|
capfd.readouterr()
|
|
assert out_file.stat().st_mode & 0o777 == 0o600
|
|
config = json.loads(out_file.read_text())
|
|
assert config["dsn"] == "AC000W00MOCK0001"
|
|
|
|
|
|
async def test_cli_monitor(mock_ac, ac_env, capfd):
|
|
proc = mock_ac()
|
|
rc = await asyncio.to_thread(
|
|
main,
|
|
[
|
|
"monitor",
|
|
"--host", "127.0.0.1",
|
|
"--device-port", str(proc.port),
|
|
"--dsn", ac_env["dsn"],
|
|
"--lanip-key", ac_env["key"],
|
|
"--lanip-key-id", str(ac_env["key_id"]),
|
|
"--listen-port", str(free_port()),
|
|
"--duration", "2",
|
|
"--get", "OperationMode",
|
|
"--log-level", "3",
|
|
],
|
|
)
|
|
assert rc == 0
|
|
out, _ = capfd.readouterr()
|
|
assert "STATE online" in out
|
|
assert "PROP OPERATION_MODE = 6" in out
|