petr.polezhaev
26d1d6d64f
- custom_components/fglair: manifest (config_flow, local_push, requirements),
const, trial (start -> ONLINE <=10 c -> GET базовых свойств), FglairClient
(сессия pyfglair в executor, колбэки в loop, --runtime_data),
__init__ (setup/unload, рефкаунт глобального лог-пампа)
- config flow: меню cloud/manual/import_json + import-source; discover через
pyfglair.provision; пробное подключение с диагностикой (cannot_connect,
no_slots, key_mismatch, timeout, invalid_auth, invalid_import); unique_id=dsn
- strings.json + translations/{en,ru}.json
- тесты HA (pytest-homeassistant-custom-component): меню/шаги/ошибки/импорт/
облако (discover замокан), setup/unload, занятый порт; trial против
mock_ac.py (online/key_mismatch/unreachable/no_slots)
- pytest.ini (общий), tests/conftest перенесён на уровень tests/,
scripts/py-ci.sh: --setup-ha и прогон tests/components
77 lines
2.0 KiB
Python
77 lines
2.0 KiB
Python
"""Пробное подключение (trial_connect) против mock_ac.py — без HA-рантайма."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
from helpers import free_port
|
|
|
|
from custom_components.fglair.trial import trial_connect
|
|
from pyfglair import Prop, State
|
|
|
|
|
|
async def test_trial_online(mock_ac, ac_env, socket_enabled):
|
|
proc = mock_ac()
|
|
result = await asyncio.to_thread(
|
|
trial_connect,
|
|
"127.0.0.1",
|
|
ac_env["dsn"],
|
|
ac_env["key"],
|
|
ac_env["key_id"],
|
|
device_port=proc.port,
|
|
listen_port=free_port(),
|
|
timeout=10.0,
|
|
)
|
|
assert result.ok, result.reason
|
|
assert result.state == State.ONLINE
|
|
assert result.properties[Prop.OPERATION_MODE].int_value == 6
|
|
assert result.properties[Prop.FAN_SPEED].int_value == 4
|
|
|
|
|
|
async def test_trial_key_mismatch(mock_ac, ac_env, socket_enabled):
|
|
proc = mock_ac()
|
|
result = await asyncio.to_thread(
|
|
trial_connect,
|
|
"127.0.0.1",
|
|
ac_env["dsn"],
|
|
ac_env["key"],
|
|
ac_env["key_id"] + 1,
|
|
device_port=proc.port,
|
|
listen_port=free_port(),
|
|
timeout=10.0,
|
|
)
|
|
assert not result.ok
|
|
assert result.reason == "key_mismatch"
|
|
|
|
|
|
async def test_trial_unreachable(ac_env, socket_enabled):
|
|
dead_port = free_port()
|
|
result = await asyncio.to_thread(
|
|
trial_connect,
|
|
"127.0.0.1",
|
|
ac_env["dsn"],
|
|
ac_env["key"],
|
|
ac_env["key_id"],
|
|
device_port=dead_port,
|
|
listen_port=free_port(),
|
|
timeout=1.0,
|
|
)
|
|
assert not result.ok
|
|
assert result.reason in ("unreachable", "timeout")
|
|
assert result.state != State.ONLINE
|
|
|
|
|
|
async def test_trial_no_slots(mock_ac, ac_env, socket_enabled):
|
|
proc = mock_ac(["--503"])
|
|
result = await asyncio.to_thread(
|
|
trial_connect,
|
|
"127.0.0.1",
|
|
ac_env["dsn"],
|
|
ac_env["key"],
|
|
ac_env["key_id"],
|
|
device_port=proc.port,
|
|
listen_port=free_port(),
|
|
timeout=3.0,
|
|
)
|
|
assert not result.ok
|
|
assert result.reason == "no_slots"
|