mirror of
https://github.com/esphome/esphome.git
synced 2026-02-08 08:41:59 +00:00
252 lines
9.2 KiB
Python
252 lines
9.2 KiB
Python
"""Test scheduler numeric ID (uint32_t) overloads."""
|
|
|
|
import asyncio
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scheduler_numeric_id_test(
|
|
yaml_config: str,
|
|
run_compiled: RunCompiledFunction,
|
|
api_client_connected: APIClientConnectedFactory,
|
|
) -> None:
|
|
"""Test that scheduler handles numeric IDs (uint32_t) correctly."""
|
|
# Track counts
|
|
timeout_count = 0
|
|
interval_count = 0
|
|
retry_count = 0
|
|
defer_count = 0
|
|
|
|
# Events for each test completion
|
|
numeric_timeout_1001_fired = asyncio.Event()
|
|
numeric_timeout_1002_fired = asyncio.Event()
|
|
numeric_interval_2001_fired = asyncio.Event()
|
|
numeric_interval_cancelled = asyncio.Event()
|
|
numeric_timeout_cancelled = asyncio.Event()
|
|
duplicate_timeout_fired = asyncio.Event()
|
|
component_timeout_fired = asyncio.Event()
|
|
component_interval_fired = asyncio.Event()
|
|
zero_id_timeout_fired = asyncio.Event()
|
|
max_id_timeout_fired = asyncio.Event()
|
|
numeric_retry_done = asyncio.Event()
|
|
numeric_retry_cancelled = asyncio.Event()
|
|
numeric_defer_7001_fired = asyncio.Event()
|
|
numeric_defer_7002_fired = asyncio.Event()
|
|
numeric_defer_cancelled = asyncio.Event()
|
|
final_results_logged = asyncio.Event()
|
|
|
|
# Track interval counts
|
|
numeric_interval_count = 0
|
|
numeric_retry_count = 0
|
|
|
|
def on_log_line(line: str) -> None:
|
|
nonlocal timeout_count, interval_count, retry_count, defer_count
|
|
nonlocal numeric_interval_count, numeric_retry_count
|
|
|
|
# Strip ANSI color codes
|
|
clean_line = re.sub(r"\x1b\[[0-9;]*m", "", line)
|
|
|
|
# Check for numeric timeout completions
|
|
if "Numeric timeout 1001 fired" in clean_line:
|
|
numeric_timeout_1001_fired.set()
|
|
timeout_count += 1
|
|
|
|
elif "Numeric timeout 1002 fired" in clean_line:
|
|
numeric_timeout_1002_fired.set()
|
|
timeout_count += 1
|
|
|
|
# Check for numeric interval
|
|
elif "Numeric interval 2001 fired" in clean_line:
|
|
match = re.search(r"count: (\d+)", clean_line)
|
|
if match:
|
|
numeric_interval_count = int(match.group(1))
|
|
numeric_interval_2001_fired.set()
|
|
|
|
elif "Cancelled numeric interval 2001" in clean_line:
|
|
numeric_interval_cancelled.set()
|
|
|
|
elif "Cancelled numeric timeout 3001" in clean_line:
|
|
numeric_timeout_cancelled.set()
|
|
|
|
# Check for duplicate timeout (only last should fire)
|
|
elif "Duplicate numeric timeout" in clean_line:
|
|
match = re.search(r"timeout (\d+) fired", clean_line)
|
|
if match and match.group(1) == "4":
|
|
duplicate_timeout_fired.set()
|
|
timeout_count += 1
|
|
|
|
# Check for component method tests
|
|
elif "Component numeric timeout 5001 fired" in clean_line:
|
|
component_timeout_fired.set()
|
|
timeout_count += 1
|
|
|
|
elif "Component numeric interval 5002 fired" in clean_line:
|
|
component_interval_fired.set()
|
|
interval_count += 1
|
|
|
|
# Check for edge case tests
|
|
elif "Numeric timeout with ID 0 fired" in clean_line:
|
|
zero_id_timeout_fired.set()
|
|
timeout_count += 1
|
|
|
|
elif "Numeric timeout with max ID fired" in clean_line:
|
|
max_id_timeout_fired.set()
|
|
timeout_count += 1
|
|
|
|
# Check for numeric retry tests
|
|
elif "Numeric retry 6001 attempt" in clean_line:
|
|
match = re.search(r"attempt (\d+)", clean_line)
|
|
if match:
|
|
numeric_retry_count = int(match.group(1))
|
|
|
|
elif "Numeric retry 6001 done" in clean_line:
|
|
numeric_retry_done.set()
|
|
|
|
elif "Cancelled numeric retry 6002" in clean_line:
|
|
numeric_retry_cancelled.set()
|
|
|
|
# Check for numeric defer tests
|
|
elif "Component numeric defer 7001 fired" in clean_line:
|
|
numeric_defer_7001_fired.set()
|
|
|
|
elif "Component numeric defer 7002 fired" in clean_line:
|
|
numeric_defer_7002_fired.set()
|
|
|
|
elif "Cancelled numeric defer 8001: true" in clean_line:
|
|
numeric_defer_cancelled.set()
|
|
|
|
# Check for final results
|
|
elif "Final results" in clean_line:
|
|
match = re.search(
|
|
r"Timeouts: (\d+), Intervals: (\d+), Retries: (\d+), Defers: (\d+)",
|
|
clean_line,
|
|
)
|
|
if match:
|
|
timeout_count = int(match.group(1))
|
|
interval_count = int(match.group(2))
|
|
retry_count = int(match.group(3))
|
|
defer_count = int(match.group(4))
|
|
final_results_logged.set()
|
|
|
|
async with (
|
|
run_compiled(yaml_config, line_callback=on_log_line),
|
|
api_client_connected() as client,
|
|
):
|
|
# Verify we can connect
|
|
device_info = await client.device_info()
|
|
assert device_info is not None
|
|
assert device_info.name == "scheduler-numeric-id-test"
|
|
|
|
# Wait for numeric timeout tests
|
|
try:
|
|
await asyncio.wait_for(numeric_timeout_1001_fired.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Numeric timeout 1001 did not fire within 0.5 seconds")
|
|
|
|
try:
|
|
await asyncio.wait_for(numeric_timeout_1002_fired.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Numeric timeout 1002 did not fire within 0.5 seconds")
|
|
|
|
try:
|
|
await asyncio.wait_for(numeric_interval_2001_fired.wait(), timeout=1.0)
|
|
except TimeoutError:
|
|
pytest.fail("Numeric interval 2001 did not fire within 1 second")
|
|
|
|
try:
|
|
await asyncio.wait_for(numeric_interval_cancelled.wait(), timeout=2.0)
|
|
except TimeoutError:
|
|
pytest.fail("Numeric interval 2001 was not cancelled within 2 seconds")
|
|
|
|
# Verify numeric interval ran at least twice
|
|
assert numeric_interval_count >= 2, (
|
|
f"Expected numeric interval to run at least 2 times, got {numeric_interval_count}"
|
|
)
|
|
|
|
# Verify numeric timeout was cancelled
|
|
assert numeric_timeout_cancelled.is_set(), (
|
|
"Numeric timeout 3001 should have been cancelled"
|
|
)
|
|
|
|
# Wait for duplicate timeout (only last one should fire)
|
|
try:
|
|
await asyncio.wait_for(duplicate_timeout_fired.wait(), timeout=1.0)
|
|
except TimeoutError:
|
|
pytest.fail("Duplicate numeric timeout did not fire within 1 second")
|
|
|
|
# Wait for component method tests
|
|
try:
|
|
await asyncio.wait_for(component_timeout_fired.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Component numeric timeout did not fire within 0.5 seconds")
|
|
|
|
try:
|
|
await asyncio.wait_for(component_interval_fired.wait(), timeout=1.0)
|
|
except TimeoutError:
|
|
pytest.fail("Component numeric interval did not fire within 1 second")
|
|
|
|
# Wait for edge case tests
|
|
try:
|
|
await asyncio.wait_for(zero_id_timeout_fired.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Zero ID timeout did not fire within 0.5 seconds")
|
|
|
|
try:
|
|
await asyncio.wait_for(max_id_timeout_fired.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Max ID timeout did not fire within 0.5 seconds")
|
|
|
|
# Wait for numeric retry tests
|
|
try:
|
|
await asyncio.wait_for(numeric_retry_done.wait(), timeout=1.0)
|
|
except TimeoutError:
|
|
pytest.fail(
|
|
f"Numeric retry 6001 did not complete. Count: {numeric_retry_count}"
|
|
)
|
|
|
|
assert numeric_retry_count >= 2, (
|
|
f"Expected at least 2 numeric retry attempts, got {numeric_retry_count}"
|
|
)
|
|
|
|
# Verify numeric retry was cancelled
|
|
assert numeric_retry_cancelled.is_set(), (
|
|
"Numeric retry 6002 should have been cancelled"
|
|
)
|
|
|
|
# Wait for numeric defer tests
|
|
try:
|
|
await asyncio.wait_for(numeric_defer_7001_fired.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Numeric defer 7001 did not fire within 0.5 seconds")
|
|
|
|
try:
|
|
await asyncio.wait_for(numeric_defer_7002_fired.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Numeric defer 7002 did not fire within 0.5 seconds")
|
|
|
|
# Verify numeric defer was cancelled
|
|
try:
|
|
await asyncio.wait_for(numeric_defer_cancelled.wait(), timeout=0.5)
|
|
except TimeoutError:
|
|
pytest.fail("Numeric defer 8001 cancel confirmation not received")
|
|
|
|
# Wait for final results
|
|
try:
|
|
await asyncio.wait_for(final_results_logged.wait(), timeout=3.0)
|
|
except TimeoutError:
|
|
pytest.fail("Final results were not logged within 3 seconds")
|
|
|
|
# Verify results
|
|
assert timeout_count >= 6, f"Expected at least 6 timeouts, got {timeout_count}"
|
|
assert interval_count >= 3, (
|
|
f"Expected at least 3 interval fires, got {interval_count}"
|
|
)
|
|
assert retry_count >= 2, (
|
|
f"Expected at least 2 retry attempts, got {retry_count}"
|
|
)
|
|
assert defer_count >= 2, f"Expected at least 2 defer fires, got {defer_count}"
|