mirror of
https://github.com/esphome/esphome.git
synced 2025-09-02 11:22:24 +01:00
92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
"""Test ESPHome automations functionality."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from .types import APIClientConnectedFactory, RunCompiledFunction
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delay_action_cancellation(
|
|
yaml_config: str,
|
|
run_compiled: RunCompiledFunction,
|
|
api_client_connected: APIClientConnectedFactory,
|
|
) -> None:
|
|
"""Test that delay actions can be properly cancelled when script restarts."""
|
|
loop = asyncio.get_running_loop()
|
|
|
|
# Track log messages with timestamps
|
|
log_entries: list[tuple[float, str]] = []
|
|
script_starts: list[float] = []
|
|
delay_completions: list[float] = []
|
|
script_restart_logged = False
|
|
test_started_time = None
|
|
|
|
# Patterns to match
|
|
test_start_pattern = re.compile(r"Starting first script execution")
|
|
script_start_pattern = re.compile(r"Script started, beginning delay")
|
|
restart_pattern = re.compile(r"Restarting script \(should cancel first delay\)")
|
|
delay_complete_pattern = re.compile(r"Delay completed successfully")
|
|
|
|
# Future to track when we can check results
|
|
second_script_started = loop.create_future()
|
|
|
|
def check_output(line: str) -> None:
|
|
"""Check log output for expected messages."""
|
|
nonlocal script_restart_logged, test_started_time
|
|
|
|
current_time = loop.time()
|
|
log_entries.append((current_time, line))
|
|
|
|
if test_start_pattern.search(line):
|
|
test_started_time = current_time
|
|
elif script_start_pattern.search(line) and test_started_time:
|
|
script_starts.append(current_time)
|
|
if len(script_starts) == 2 and not second_script_started.done():
|
|
second_script_started.set_result(True)
|
|
elif restart_pattern.search(line):
|
|
script_restart_logged = True
|
|
elif delay_complete_pattern.search(line):
|
|
delay_completions.append(current_time)
|
|
|
|
async with (
|
|
run_compiled(yaml_config, line_callback=check_output),
|
|
api_client_connected() as client,
|
|
):
|
|
# Get services
|
|
entities, services = await client.list_entities_services()
|
|
|
|
# Find our test service
|
|
test_service = next(
|
|
(s for s in services if s.name == "start_delay_then_restart"), None
|
|
)
|
|
assert test_service is not None, "start_delay_then_restart service not found"
|
|
|
|
# Execute the test sequence
|
|
client.execute_service(test_service, {})
|
|
|
|
# Wait for the second script to start
|
|
await asyncio.wait_for(second_script_started, timeout=5.0)
|
|
|
|
# Wait for potential delay completion
|
|
await asyncio.sleep(0.75) # Original delay was 500ms
|
|
|
|
# Check results
|
|
assert len(script_starts) == 2, (
|
|
f"Script should have started twice, but started {len(script_starts)} times"
|
|
)
|
|
assert script_restart_logged, "Script restart was not logged"
|
|
|
|
# Verify we got exactly one completion and it happened ~500ms after the second start
|
|
assert len(delay_completions) == 1, (
|
|
f"Expected 1 delay completion, got {len(delay_completions)}"
|
|
)
|
|
time_from_second_start = delay_completions[0] - script_starts[1]
|
|
assert 0.4 < time_from_second_start < 0.6, (
|
|
f"Delay completed {time_from_second_start:.3f}s after second start, expected ~0.5s"
|
|
)
|