Home/ClimatController
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

work: Автоматический режим управления скоростью

Browse Source 
отлажен!
This commit is contained in:
Fedorov Dmitriy
2026-09-19 15:48:44 +03:00
parent cdec9b0996
commit f98cc2d6a3
6 changed files with 867 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_qingping_recovery.py
+415
View File
@@ -0,0 +1,415 @@
import json
from types import SimpleNamespace
from unittest.mock import patch
import paho.mqtt.client as mqtt
from app.my_dataclasses import (
QINGPING_SAMPLE_TIMEOUT,
QINGPING_RECOVERY_TIMEOUT,
)
from app.qingping.service import QingpingService
class FakeClock:
def __init__(self):
self.now = 0.0
def monotonic(self) -> float:
return self.now
def advance(self, seconds: float) -> None:
self.now += seconds
class FakeMqttClient:
def __init__(self):
self.subscriptions = []
self.published = []
def subscribe(self, topic):
self.subscriptions.append(topic)
def publish(
self,
topic,
payload,
):
self.published.append(
(
topic,
json.loads(payload),
)
)
return SimpleNamespace(
rc=mqtt.MQTT_ERR_SUCCESS
)
class FakeMessage:
def __init__(self, payload: dict):
self.payload = json.dumps(
payload
).encode("utf-8")
def sensor_message(
timestamp: int,
co2: int,
) -> FakeMessage:
return FakeMessage(
{
"type": 17,
"sensorData": [
{
"timestamp": {
"value": timestamp
},
"temperature": {
"value": 25.0
},
"humidity": {
"value": 50.0
},
"co2": {
"value": co2
},
"pm25": {
"value": 1
},
"pm10": {
"value": 1
},
"battery": {
"value": 100
},
}
],
}
)
def service_message() -> FakeMessage:
return FakeMessage(
{
"type": 13,
"wifi_info": (
"TestWiFi,-50,1,"
"00:00:00:00:00:00"
),
"sw_version": "4.8.5",
}
)
def assert_state(
service: QingpingService,
*,
online: bool,
recovery_waiting: bool,
co2,
):
status = service.status()
assert status["online"] is online, status
assert (
status["recovery_waiting"]
is recovery_waiting
), status
assert status["co2"] == co2, status
def test_double_recovery_cycle() -> None:
print()
print("TEST — DOUBLE RECOVERY CYCLE")
clock = FakeClock()
service = QingpingService(
host="127.0.0.1",
)
client = FakeMqttClient()
# _send_recovery() использует self._client.
service._client = client
with patch(
"app.qingping.service.time.monotonic",
clock.monotonic,
):
# --------------------------------------------------
# START
# --------------------------------------------------
print("1. MQTT connect")
service._on_connect(
client,
None,
None,
0,
None,
)
assert len(client.published) == 1
assert_state(
service,
online=False,
recovery_waiting=True,
co2=None,
)
# --------------------------------------------------
# Первый type 17.
# --------------------------------------------------
print("2. First sensor sample")
service._on_message(
None,
None,
sensor_message(
timestamp=1000,
co2=900,
),
)
assert_state(
service,
online=True,
recovery_waiting=False,
co2=900,
)
# --------------------------------------------------
# Первый отказ.
# --------------------------------------------------
print("3. First sensor timeout")
clock.advance(
QINGPING_SAMPLE_TIMEOUT + 1
)
service._watchdog_tick(
clock.monotonic()
)
assert_state(
service,
online=False,
recovery_waiting=False,
co2=900,
)
# Recovery пока НЕ отправлялся.
assert len(client.published) == 1
# --------------------------------------------------
# Устройство снова появляется.
# Любой пакет запускает recovery.
# --------------------------------------------------
print("4. First recovery")
service._on_message(
None,
None,
service_message(),
)
assert len(client.published) == 2
assert_state(
service,
online=True,
recovery_waiting=True,
co2=None,
)
# --------------------------------------------------
# Новый type 17.
# --------------------------------------------------
print("5. First recovery completed")
service._on_message(
None,
None,
sensor_message(
timestamp=1015,
co2=1000,
),
)
assert_state(
service,
online=True,
recovery_waiting=False,
co2=1000,
)
# --------------------------------------------------
# Второй отказ.
# --------------------------------------------------
print("6. Second sensor timeout")
clock.advance(
QINGPING_SAMPLE_TIMEOUT + 1
)
service._watchdog_tick(
clock.monotonic()
)
assert_state(
service,
online=False,
recovery_waiting=False,
co2=1000,
)
assert len(client.published) == 2
# --------------------------------------------------
# Второе восстановление.
# --------------------------------------------------
print("7. Second recovery")
service._on_message(
None,
None,
service_message(),
)
assert len(client.published) == 3
assert_state(
service,
online=True,
recovery_waiting=True,
co2=None,
)
service._on_message(
None,
None,
sensor_message(
timestamp=1030,
co2=1100,
),
)
assert_state(
service,
online=True,
recovery_waiting=False,
co2=1100,
)
print("8. Second recovery completed")
def test_recovery_timeout_and_retry() -> None:
print()
print("TEST — RECOVERY TIMEOUT AND RETRY")
clock = FakeClock()
service = QingpingService(
host="127.0.0.1",
)
client = FakeMqttClient()
service._client = client
with patch(
"app.qingping.service.time.monotonic",
clock.monotonic,
):
# Startup recovery.
service._on_connect(
client,
None,
None,
0,
None,
)
assert len(client.published) == 1
assert_state(
service,
online=False,
recovery_waiting=True,
co2=None,
)
# Type 17 так и не пришёл.
clock.advance(
QINGPING_RECOVERY_TIMEOUT + 1
)
service._watchdog_tick(
clock.monotonic()
)
assert_state(
service,
online=False,
recovery_waiting=False,
co2=None,
)
# Любой следующий пакет должен разрешить
# новую recovery-попытку.
service._on_message(
None,
None,
service_message(),
)
assert len(client.published) == 2
assert_state(
service,
online=True,
recovery_waiting=True,
co2=None,
)
# Теперь приходит type 17.
service._on_message(
None,
None,
sensor_message(
timestamp=2000,
co2=1200,
),
)
assert_state(
service,
online=True,
recovery_waiting=False,
co2=1200,
)
if __name__ == "__main__":
test_double_recovery_cycle()
test_recovery_timeout_and_retry()
print()
print(
"ALL QINGPING RECOVERY TESTS PASSED"
)