github/esphome
2
0
Fork 0
mirror of https://github.com/esphome/esphome.git synced 2026-10-21 21:36:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

[tests] Use require_service in the integration tests (#21214)

Browse Source
This commit is contained in:
J. Nick Koston
2026-10-21 01:47:31 +00:00
committed by GitHub
parent f47dca7e4e
commit 7c58a9698b
44 changed files with 133 additions and 337 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_api_action_responses.py
+7 -27
View File
@@ -9,10 +9,10 @@ import asyncio
import json
import re
from aioesphomeapi import SupportsResponseType, UserService, UserServiceArgType
from aioesphomeapi import SupportsResponseType, UserServiceArgType
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -79,31 +79,11 @@ async def test_api_action_responses(
assert len(services) == 5, f"Expected 5 services, found {len(services)}"
# Find our services
action_no_response: UserService | None = None
action_status_response: UserService | None = None
action_optional_response: UserService | None = None
action_only_response: UserService | None = None
action_nested_json: UserService | None = None
for service in services:
if service.name == "action_no_response":
action_no_response = service
elif service.name == "action_status_response":
action_status_response = service
elif service.name == "action_optional_response":
action_optional_response = service
elif service.name == "action_only_response":
action_only_response = service
elif service.name == "action_nested_json":
action_nested_json = service
assert action_no_response is not None, "action_no_response not found"
assert action_status_response is not None, "action_status_response not found"
assert action_optional_response is not None, (
"action_optional_response not found"
)
assert action_only_response is not None, "action_only_response not found"
assert action_nested_json is not None, "action_nested_json not found"
action_no_response = require_service(services, "action_no_response")
action_status_response = require_service(services, "action_status_response")
action_optional_response = require_service(services, "action_optional_response")
action_only_response = require_service(services, "action_only_response")
action_nested_json = require_service(services, "action_nested_json")
# Verify supports_response modes
assert action_no_response.supports_response is None or (
tests/integration/test_api_action_timeout.py
+4 -17
View File
@@ -10,10 +10,9 @@ import asyncio
import contextlib
import re
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -89,21 +88,9 @@ async def test_api_action_timeout(
assert len(services) == 3, f"Expected 3 services, found {len(services)}"
# Find our services
action_immediate: UserService | None = None
action_short_delay: UserService | None = None
action_long_delay: UserService | None = None
for service in services:
if service.name == "action_immediate":
action_immediate = service
elif service.name == "action_short_delay":
action_short_delay = service
elif service.name == "action_long_delay":
action_long_delay = service
assert action_immediate is not None, "action_immediate not found"
assert action_short_delay is not None, "action_short_delay not found"
assert action_long_delay is not None, "action_long_delay not found"
action_immediate = require_service(services, "action_immediate")
action_short_delay = require_service(services, "action_short_delay")
action_long_delay = require_service(services, "action_long_delay")
# Test 1: Immediate response should work
response = await client.execute_service(
tests/integration/test_api_conditional_memory.py
+4 -13
View File
@@ -5,10 +5,10 @@ from __future__ import annotations
import asyncio
import re
from aioesphomeapi import UserService, UserServiceArgType
from aioesphomeapi import UserServiceArgType
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -66,17 +66,8 @@ async def test_api_conditional_memory(
assert len(services) == 2, f"Expected 2 services, found {len(services)}"
# Find our services
simple_service: UserService | None = None
service_with_args: UserService | None = None
for service in services:
if service.name == "test_simple_service":
simple_service = service
elif service.name == "test_service_with_args":
service_with_args = service
assert simple_service is not None, "test_simple_service not found"
assert service_with_args is not None, "test_service_with_args not found"
simple_service = require_service(services, "test_simple_service")
service_with_args = require_service(services, "test_service_with_args")
# Verify service arguments
assert len(service_with_args.args) == 4, (
tests/integration/test_api_custom_services.py
+9 -30
View File
@@ -5,10 +5,10 @@ from __future__ import annotations
import asyncio
import re
from aioesphomeapi import UserService, UserServiceArgType
from aioesphomeapi import UserServiceArgType
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -77,35 +77,14 @@ async def test_api_custom_services(
assert len(services) == 6, f"Expected 6 services, found {len(services)}"
# Find our services
yaml_service: UserService | None = None
yaml_args_service: UserService | None = None
yaml_many_args_service: UserService | None = None
custom_service: UserService | None = None
custom_args_service: UserService | None = None
custom_arrays_service: UserService | None = None
for service in services:
if service.name == "test_yaml_service":
yaml_service = service
elif service.name == "test_yaml_service_with_args":
yaml_args_service = service
elif service.name == "test_yaml_service_many_args":
yaml_many_args_service = service
elif service.name == "custom_test_service":
custom_service = service
elif service.name == "custom_service_with_args":
custom_args_service = service
elif service.name == "custom_service_with_arrays":
custom_arrays_service = service
assert yaml_service is not None, "test_yaml_service not found"
assert yaml_args_service is not None, "test_yaml_service_with_args not found"
assert yaml_many_args_service is not None, (
"test_yaml_service_many_args not found"
yaml_service = require_service(services, "test_yaml_service")
yaml_args_service = require_service(services, "test_yaml_service_with_args")
yaml_many_args_service = require_service(
services, "test_yaml_service_many_args"
)
assert custom_service is not None, "custom_test_service not found"
assert custom_args_service is not None, "custom_service_with_args not found"
assert custom_arrays_service is not None, "custom_service_with_arrays not found"
custom_service = require_service(services, "custom_test_service")
custom_args_service = require_service(services, "custom_service_with_args")
custom_arrays_service = require_service(services, "custom_service_with_arrays")
# Test YAML service
await client.execute_service(yaml_service, {})
tests/integration/test_api_homeassistant.py
+2 -5
View File
@@ -16,7 +16,7 @@ import re
from aioesphomeapi import HomeassistantServiceCall
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -227,10 +227,7 @@ async def test_api_homeassistant(
_, services = await client.list_entities_services()
# Find the trigger service
trigger_service = next(
(s for s in services if s.name == "trigger_all_tests"), None
)
assert trigger_service is not None, "trigger_all_tests service not found"
trigger_service = require_service(services, "trigger_all_tests")
# Execute all tests
await client.execute_service(trigger_service, {})
tests/integration/test_api_homeassistant_entities.py
+2 -1
View File
@@ -9,6 +9,7 @@ from aioesphomeapi import APIClient, HomeassistantServiceCall, UserService
import pytest
from .log_utils import LineWaiter
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -73,7 +74,7 @@ async def test_api_homeassistant_entities(
):
entities, services = await client.list_entities_services()
keys = {entity.name: entity.key for entity in entities}
dump_select = next(s for s in services if s.name == "dump_select")
dump_select = require_service(services, "dump_select")
client.subscribe_service_calls(calls.put_nowait)
client.subscribe_home_assistant_states(on_state_sub)
await wait_or_fail(
tests/integration/test_api_homeassistant_light.py
+2 -2
View File
@@ -21,7 +21,7 @@ from google.protobuf import message
import pytest
from .log_utils import LineWaiter
from .state_utils import require_entity
from .state_utils import require_entity, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
ENTITY = "light.kitchen"
@@ -160,7 +160,7 @@ async def test_api_homeassistant_light(
async with light_entity(
yaml_config, run_compiled, api_client_connected, "['color_temp', 'rgb']"
) as ha:
set_force_off = next(s for s in ha.services if s.name == "set_force_off")
set_force_off = require_service(ha.services, "set_force_off")
# Normal while Home Assistant starts, so no warning
ha.report(None, "unavailable")
ha.report(None, "unknown")
tests/integration/test_api_string_lambda.py
+5 -15
View File
@@ -7,7 +7,7 @@ import re
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -56,23 +56,13 @@ async def test_api_string_lambda(
_, services = await client.list_entities_services()
# Find all test services
string_service = next(
(s for s in services if s.name == "test_string_lambda"), None
)
assert string_service is not None, "test_string_lambda service not found"
string_service = require_service(services, "test_string_lambda")
int_service = next((s for s in services if s.name == "test_int_lambda"), None)
assert int_service is not None, "test_int_lambda service not found"
int_service = require_service(services, "test_int_lambda")
float_service = next(
(s for s in services if s.name == "test_float_lambda"), None
)
assert float_service is not None, "test_float_lambda service not found"
float_service = require_service(services, "test_float_lambda")
char_ptr_service = next(
(s for s in services if s.name == "test_char_ptr_lambda"), None
)
assert char_ptr_service is not None, "test_char_ptr_lambda service not found"
char_ptr_service = require_service(services, "test_char_ptr_lambda")
# Execute all four services to test different lambda return types
await client.execute_service(
tests/integration/test_automation_wait_actions.py
+4 -8
View File
@@ -7,6 +7,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -69,8 +70,7 @@ async def test_automation_wait_actions(
_, services = await client.list_entities_services()
# Test 1: wait_until in automation - trigger 5 times rapidly
test_service = next((s for s in services if s.name == "test_wait_until"), None)
assert test_service is not None, "test_wait_until service not found"
test_service = require_service(services, "test_wait_until")
await client.execute_service(test_service, {})
await asyncio.wait_for(test1_complete, timeout=3.0)
@@ -80,8 +80,7 @@ async def test_automation_wait_actions(
)
# Test 2: script.wait in automation - trigger 5 times rapidly
test_service = next((s for s in services if s.name == "test_script_wait"), None)
assert test_service is not None, "test_script_wait service not found"
test_service = require_service(services, "test_script_wait")
await client.execute_service(test_service, {})
await asyncio.wait_for(test2_complete, timeout=3.0)
@@ -91,10 +90,7 @@ async def test_automation_wait_actions(
)
# Test 3: wait_until with timeout in automation - trigger 5 times rapidly
test_service = next(
(s for s in services if s.name == "test_wait_timeout"), None
)
assert test_service is not None, "test_wait_timeout service not found"
test_service = require_service(services, "test_wait_timeout")
await client.execute_service(test_service, {})
await asyncio.wait_for(test3_complete, timeout=3.0)
tests/integration/test_automations.py
+3 -8
View File
@@ -7,6 +7,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -61,10 +62,7 @@ async def test_delay_action_cancellation(
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"
test_service = require_service(services, "start_delay_then_restart")
# Execute the test sequence
await client.execute_service(test_service, {})
@@ -132,10 +130,7 @@ async def test_parallel_script_delays(
entities, services = await client.list_entities_services()
# Find our test service
test_service = next(
(s for s in services if s.name == "test_parallel_delays"), None
)
assert test_service is not None, "test_parallel_delays service not found"
test_service = require_service(services, "test_parallel_delays")
# Execute the test - this will start 3 parallel scripts with 1 second delays
await client.execute_service(test_service, {})
tests/integration/test_continuation_actions.py
+7 -16
View File
@@ -7,6 +7,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -137,8 +138,7 @@ async def test_continuation_actions(
_, services = await client.list_entities_services()
# Test 1: IfAction with then branch
test_service = next((s for s in services if s.name == "test_if_action"), None)
assert test_service is not None, "test_if_action service not found"
test_service = require_service(services, "test_if_action")
await client.execute_service(test_service, {"condition": True, "value": 42})
await asyncio.wait_for(test1_complete, timeout=2.0)
assert test_results["if_then"], "IfAction then branch not executed"
@@ -153,8 +153,7 @@ async def test_continuation_actions(
assert test_results["if_complete"], "IfAction did not complete"
# Test 2: Nested IfAction - test all branches
test_service = next((s for s in services if s.name == "test_nested_if"), None)
assert test_service is not None, "test_nested_if service not found"
test_service = require_service(services, "test_nested_if")
# Both true
await client.execute_service(test_service, {"outer": True, "inner": True})
@@ -178,10 +177,7 @@ async def test_continuation_actions(
assert test_results["nested_outer_false"], "Nested outer false not executed"
# Test 3: WhileAction
test_service = next(
(s for s in services if s.name == "test_while_action"), None
)
assert test_service is not None, "test_while_action service not found"
test_service = require_service(services, "test_while_action")
await client.execute_service(test_service, {"max_count": 3})
await asyncio.wait_for(test3_complete, timeout=2.0)
assert test_results["while_iterations"] == 3, (
@@ -190,10 +186,7 @@ async def test_continuation_actions(
assert test_results["while_complete"], "WhileAction did not complete"
# Test 4: RepeatAction
test_service = next(
(s for s in services if s.name == "test_repeat_action"), None
)
assert test_service is not None, "test_repeat_action service not found"
test_service = require_service(services, "test_repeat_action")
await client.execute_service(test_service, {"count": 5})
await asyncio.wait_for(test4_complete, timeout=2.0)
assert test_results["repeat_iterations"] == 5, (
@@ -202,8 +195,7 @@ async def test_continuation_actions(
assert test_results["repeat_complete"], "RepeatAction did not complete"
# Test 5: Combined (if + repeat + while)
test_service = next((s for s in services if s.name == "test_combined"), None)
assert test_service is not None, "test_combined service not found"
test_service = require_service(services, "test_combined")
await client.execute_service(test_service, {"do_loop": True, "loop_count": 2})
await asyncio.wait_for(test5_complete, timeout=2.0)
# Should execute: repeat 2 times, each iteration does while from iteration down to 0
@@ -216,8 +208,7 @@ async def test_continuation_actions(
assert test_results["combined_complete"], "Combined did not complete"
# Test 6: Rapid triggers through an if/else
test_service = next((s for s in services if s.name == "test_rapid_if"), None)
assert test_service is not None, "test_rapid_if service not found"
test_service = require_service(services, "test_rapid_if")
await client.execute_service(test_service, {})
await asyncio.wait_for(test6_complete, timeout=2.0)
# Values 1, 2 should hit else (<=2), values 3, 4, 5 should hit then (>2)
tests/integration/test_host_write_after_peer_close.py
+2 -1
View File
@@ -12,6 +12,7 @@ from collections.abc import Callable
import pytest
from .log_utils import LineWaiter
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -39,7 +40,7 @@ async def test_host_write_after_peer_close(
):
peer = await asyncio.wait_for(peers.get(), 15.0)
_, services = await client.list_entities_services()
action = next(s for s in services if s.name == "write_after_close")
action = require_service(services, "write_after_close")
await client.execute_service(action, {"port": port})
# The action opens a second connection and blocks the loop until it gets a byte.
tests/integration/test_improv_serial_uart.py
+2 -1
View File
@@ -17,6 +17,7 @@ from __future__ import annotations
import pytest
from .log_utils import LineWaiter
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
# Improv serial framing (improv_serial_component.h)
@@ -79,7 +80,7 @@ async def test_improv_serial_uart(
api_client_connected() as client,
):
_entities, services = await client.list_entities_services()
inject = next(s for s in services if s.name == "uart_inject")
inject = require_service(services, "uart_inject")
# 1. Get Current State: expect the complete current-state frame reporting
# AUTHORIZED (0x02), checksum and newline included
tests/integration/test_in_range_conditions.py
+2 -2
View File
@@ -7,6 +7,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
# Bounds in the fixture: above 0, below 10, both inclusive. A NaN state is never in range.
@@ -48,8 +49,7 @@ async def test_in_range_conditions(
api_client_connected() as client,
):
_, services = await client.list_entities_services()
service = next((s for s in services if s.name == "run_cases"), None)
assert service is not None, "run_cases action not found"
service = require_service(services, "run_cases")
await client.execute_service(service, {})
await asyncio.wait_for(done, timeout=10.0)
tests/integration/test_lvgl_color_picker_render.py
+3 -2
View File
@@ -18,6 +18,7 @@ import pytest
from .artifact_utils import keep_artifact
from .bmp_utils import Bmp, capture_when_drawn, wait_for_bmp
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
WIDTH = 680
@@ -76,8 +77,8 @@ async def test_lvgl_color_picker_render(
async with run_compiled(yaml_config), api_client_connected() as client:
_, services = await client.list_entities_services()
take_screenshot = next(s for s in services if s.name == "take_screenshot")
set_orange = next(s for s in services if s.name == "set_orange")
take_screenshot = require_service(services, "take_screenshot")
set_orange = require_service(services, "set_orange")
async def take(name: str) -> None:
await client.execute_service(take_screenshot, {"name": name})
tests/integration/test_lvgl_headless_render.py
+2 -1
View File
@@ -18,6 +18,7 @@ import pytest
from .artifact_utils import keep_artifact
from .bmp_utils import capture_when_drawn
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
WIDTH = 300
@@ -47,7 +48,7 @@ async def test_lvgl_headless_render(
async with run_compiled(yaml_config), api_client_connected() as client:
_, services = await client.list_entities_services()
service = next(s for s in services if s.name == "take_screenshot")
service = require_service(services, "take_screenshot")
async def take(name: str) -> None:
await client.execute_service(service, {"name": name})
tests/integration/test_multi_click_trigger.py
+2 -2
View File
@@ -13,6 +13,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -51,8 +52,7 @@ async def test_multi_click_trigger(
):
_entities, services = await client.list_entities_services()
test_service = next((s for s in services if s.name == "run_all_tests"), None)
assert test_service is not None, "run_all_tests service not found"
test_service = require_service(services, "run_all_tests")
# Kick off the entire test sequence (runs in YAML with delays)
await client.execute_service(test_service, {})
tests/integration/test_scheduler_bulk_cleanup.py
+2 -11
View File
@@ -3,10 +3,9 @@
import asyncio
import re
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -75,15 +74,7 @@ async def test_scheduler_bulk_cleanup(
)
# Find our test service
trigger_bulk_cleanup_service: UserService | None = None
for service in services:
if service.name == "trigger_bulk_cleanup":
trigger_bulk_cleanup_service = service
break
assert trigger_bulk_cleanup_service is not None, (
"trigger_bulk_cleanup service not found"
)
trigger_bulk_cleanup_service = require_service(services, "trigger_bulk_cleanup")
# Execute the test
await client.execute_service(trigger_bulk_cleanup_service, {})
tests/integration/test_scheduler_defer_cancel.py
+3 -10
View File
@@ -2,10 +2,10 @@
import asyncio
from aioesphomeapi import EntityState, Event, EventInfo, UserService
from aioesphomeapi import EntityState, Event, EventInfo
import pytest
from .state_utils import assert_device_name, require_entity
from .state_utils import assert_device_name, require_entity, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -35,14 +35,7 @@ async def test_scheduler_defer_cancel(
)
# Find our test service
test_defer_cancel_service: UserService | None = None
for service in services:
if service.name == "test_defer_cancel":
test_defer_cancel_service = service
assert test_defer_cancel_service is not None, (
"test_defer_cancel service not found"
)
test_defer_cancel_service = require_service(services, "test_defer_cancel")
# Get the event loop
loop = asyncio.get_running_loop()
tests/integration/test_scheduler_defer_cancel_regular.py
+2 -9
View File
@@ -2,10 +2,9 @@
import asyncio
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -49,13 +48,7 @@ async def test_scheduler_defer_cancels_regular(
)
# Find our test service
test_service: UserService | None = None
for service in services:
if service.name == "test_defer_cancels_regular":
test_service = service
break
assert test_service is not None, "test_defer_cancels_regular service not found"
test_service = require_service(services, "test_defer_cancels_regular")
# Execute the test
await client.execute_service(test_service, {})
tests/integration/test_scheduler_defer_fifo_simple.py
+4 -14
View File
@@ -2,10 +2,10 @@
import asyncio
from aioesphomeapi import EntityState, Event, EventInfo, UserService
from aioesphomeapi import EntityState, Event, EventInfo
import pytest
from .state_utils import assert_device_name, require_entity
from .state_utils import assert_device_name, require_entity, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -35,18 +35,8 @@ async def test_scheduler_defer_fifo_simple(
)
# Find our test services
test_set_timeout_service: UserService | None = None
test_defer_service: UserService | None = None
for service in services:
if service.name == "test_set_timeout":
test_set_timeout_service = service
elif service.name == "test_defer":
test_defer_service = service
assert test_set_timeout_service is not None, (
"test_set_timeout service not found"
)
assert test_defer_service is not None, "test_defer service not found"
test_set_timeout_service = require_service(services, "test_set_timeout")
test_defer_service = require_service(services, "test_defer")
# Get the event loop
loop = asyncio.get_running_loop()
tests/integration/test_scheduler_defer_stress.py
+2 -9
View File
@@ -3,10 +3,9 @@
import asyncio
import re
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -71,13 +70,7 @@ async def test_scheduler_defer_stress(
)
# Find our test service
run_stress_test_service: UserService | None = None
for service in services:
if service.name == "run_stress_test":
run_stress_test_service = service
break
assert run_stress_test_service is not None, "run_stress_test service not found"
run_stress_test_service = require_service(services, "run_stress_test")
# Call the run_stress_test service to start the test
await client.execute_service(run_stress_test_service, {})
tests/integration/test_scheduler_heap_stress.py
+2 -11
View File
@@ -3,10 +3,9 @@
import asyncio
import re
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -76,15 +75,7 @@ async def test_scheduler_heap_stress(
)
# Find our test service
run_stress_test_service: UserService | None = None
for service in services:
if service.name == "run_heap_stress_test":
run_stress_test_service = service
break
assert run_stress_test_service is not None, (
"run_heap_stress_test service not found"
)
run_stress_test_service = require_service(services, "run_heap_stress_test")
# Call the run_heap_stress_test service to start the test
await client.execute_service(run_stress_test_service, {})
tests/integration/test_scheduler_null_name.py
+2 -5
View File
@@ -5,7 +5,7 @@ import re
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -41,10 +41,7 @@ async def test_scheduler_null_name(
)
# Find our test service
test_null_name_service = next(
(s for s in services if s.name == "test_null_name"), None
)
assert test_null_name_service is not None, "test_null_name service not found"
test_null_name_service = require_service(services, "test_null_name")
# Execute the test
await client.execute_service(test_null_name_service, {})
tests/integration/test_scheduler_pool.py
+3 -4
View File
@@ -7,7 +7,7 @@ import re
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -112,10 +112,9 @@ async def test_scheduler_pool(
# Get service objects
phase_services = {
num: next(s for s in services if s.name == f"run_phase_{num}")
for num in range(1, 8)
num: require_service(services, f"run_phase_{num}") for num in range(1, 8)
}
complete_service = next(s for s in services if s.name == "run_complete")
complete_service = require_service(services, "run_complete")
try:
# Phase 1: Component lifecycle
tests/integration/test_scheduler_rapid_cancellation.py
+2 -11
View File
@@ -3,10 +3,9 @@
import asyncio
import re
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -85,15 +84,7 @@ async def test_scheduler_rapid_cancellation(
)
# Find our test service
run_test_service: UserService | None = None
for service in services:
if service.name == "run_rapid_cancellation_test":
run_test_service = service
break
assert run_test_service is not None, (
"run_rapid_cancellation_test service not found"
)
run_test_service = require_service(services, "run_rapid_cancellation_test")
# Call the service to start the test
await client.execute_service(run_test_service, {})
tests/integration/test_scheduler_recursive_timeout.py
+2 -11
View File
@@ -2,10 +2,9 @@
import asyncio
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -56,15 +55,7 @@ async def test_scheduler_recursive_timeout(
)
# Find our test service
run_test_service: UserService | None = None
for service in services:
if service.name == "run_recursive_timeout_test":
run_test_service = service
break
assert run_test_service is not None, (
"run_recursive_timeout_test service not found"
)
run_test_service = require_service(services, "run_recursive_timeout_test")
# Call the service to start the test
await client.execute_service(run_test_service, {})
tests/integration/test_scheduler_removed_item_race.py
+2 -3
View File
@@ -5,7 +5,7 @@ import re
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -76,8 +76,7 @@ async def test_scheduler_removed_item_race(
)
# Find run_test service
run_test_service = next((s for s in services if s.name == "run_test"), None)
assert run_test_service is not None, "run_test service not found"
run_test_service = require_service(services, "run_test")
# Execute the test
await client.execute_service(run_test_service, {})
tests/integration/test_scheduler_simultaneous_callbacks.py
+2 -11
View File
@@ -3,10 +3,9 @@
import asyncio
import re
from aioesphomeapi import UserService
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -75,15 +74,7 @@ async def test_scheduler_simultaneous_callbacks(
)
# Find our test service
run_test_service: UserService | None = None
for service in services:
if service.name == "run_simultaneous_callbacks_test":
run_test_service = service
break
assert run_test_service is not None, (
"run_simultaneous_callbacks_test service not found"
)
run_test_service = require_service(services, "run_simultaneous_callbacks_test")
# Call the service to start the test
await client.execute_service(run_test_service, {})
tests/integration/test_script_array_params.py
+2 -2
View File
@@ -14,6 +14,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -50,8 +51,7 @@ async def test_script_array_params(
api_client_connected() as client,
):
_, services = await client.list_entities_services()
service = next((s for s in services if s.name == "run_array_script"), None)
assert service is not None, "run_array_script service not found"
service = require_service(services, "run_array_script")
await client.execute_service(service, {})
await wait_or_fail(
tests/integration/test_script_delay_params.py
+2 -5
View File
@@ -20,7 +20,7 @@ import re
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -84,10 +84,7 @@ async def test_script_delay_with_params(
# Get services
_, services = await client.list_entities_services()
test_service = next(
(s for s in services if s.name == "test_repeat_with_delay"), None
)
assert test_service is not None, "test_repeat_with_delay service not found"
test_service = require_service(services, "test_repeat_with_delay")
# Execute the test
await client.execute_service(test_service, {})
tests/integration/test_script_queued.py
+8 -14
View File
@@ -7,6 +7,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -166,8 +167,7 @@ async def test_script_queued(
)
# Test 1: Queue depth limit
test_service = next((s for s in services if s.name == "test_queue_depth"), None)
assert test_service is not None, "test_queue_depth service not found"
test_service = require_service(services, "test_queue_depth")
await client.execute_service(test_service, {})
await asyncio.wait_for(test1_complete, timeout=2.0)
await asyncio.sleep(0.1) # Give time for rejections
@@ -181,8 +181,7 @@ async def test_script_queued(
)
# Test 2: Ring buffer order
test_service = next((s for s in services if s.name == "test_ring_buffer"), None)
assert test_service is not None, "test_ring_buffer service not found"
test_service = require_service(services, "test_ring_buffer")
await client.execute_service(test_service, {})
await asyncio.wait_for(test2_complete, timeout=2.0)
@@ -195,8 +194,7 @@ async def test_script_queued(
)
# Test 3: Stop clears queue
test_service = next((s for s in services if s.name == "test_stop_clears"), None)
assert test_service is not None, "test_stop_clears service not found"
test_service = require_service(services, "test_stop_clears")
await client.execute_service(test_service, {})
await asyncio.wait_for(test3_complete, timeout=2.0)
@@ -209,8 +207,7 @@ async def test_script_queued(
)
# Test 4: Rejection enforcement (max_runs=3)
test_service = next((s for s in services if s.name == "test_rejection"), None)
assert test_service is not None, "test_rejection service not found"
test_service = require_service(services, "test_rejection")
await client.execute_service(test_service, {})
await asyncio.wait_for(test4_complete, timeout=2.0)
await asyncio.sleep(0.1) # Give time for rejections
@@ -224,8 +221,7 @@ async def test_script_queued(
)
# Test 5: No parameters
test_service = next((s for s in services if s.name == "test_no_params"), None)
assert test_service is not None, "test_no_params service not found"
test_service = require_service(services, "test_no_params")
await client.execute_service(test_service, {})
await asyncio.wait_for(test5_complete, timeout=2.0)
@@ -236,8 +232,7 @@ async def test_script_queued(
# Test 5 again: after the queue fully drained (loop disabled while
# idle), executing again must still work
test_service = next((s for s in services if s.name == "test_no_params"), None)
assert test_service is not None, "test_no_params service not found"
test_service = require_service(services, "test_no_params")
await client.execute_service(test_service, {})
await asyncio.wait_for(test5_again_complete, timeout=2.0)
assert test_results["no_params"]["executions"] == 6, (
@@ -246,7 +241,6 @@ async def test_script_queued(
# Test 6: a stopped script (queue cleared, loop disabled) must run
# again on the next execute; the future resolves only on "END 9"
test_service = next((s for s in services if s.name == "test_after_stop"), None)
assert test_service is not None, "test_after_stop service not found"
test_service = require_service(services, "test_after_stop")
await client.execute_service(test_service, {})
await asyncio.wait_for(test6_complete, timeout=2.0)
tests/integration/test_script_queued_idle_loop.py
+2 -2
View File
@@ -7,6 +7,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -70,8 +71,7 @@ async def test_script_queued_idle_loop(
await asyncio.wait_for(boot_disabled, timeout=5.0)
_, services = await client.list_entities_services()
run_twice = next((s for s in services if s.name == "run_twice"), None)
assert run_twice is not None, "run_twice service not found"
run_twice = require_service(services, "run_twice")
await client.execute_service(run_twice, {})
# Queueing the second instance must re-enable the loop
tests/integration/test_sdl_headless_screenshot.py
+2 -1
View File
@@ -12,6 +12,7 @@ from pathlib import Path
import pytest
from .bmp_utils import capture_when_drawn
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
WIDTH = 101
@@ -37,7 +38,7 @@ async def test_sdl_headless_screenshot(
async with run_compiled(yaml_config), api_client_connected() as client:
_, services = await client.list_entities_services()
service = next(s for s in services if s.name == "take_screenshot")
service = require_service(services, "take_screenshot")
async def take(name: str) -> None:
await client.execute_service(service, {"name": name})
tests/integration/test_set_internal_at_boot.py
+2 -1
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import pytest
from .log_utils import LineWaiter
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -28,7 +29,7 @@ async def test_set_internal_at_boot(
assert "Shown At Boot" in names
assert "Untouched" in names
late = next(s for s in services if s.name == "set_internal_late")
late = require_service(services, "set_internal_late")
await client.execute_service(late, {})
await waiter.wait_for(
"'Untouched'",
tests/integration/test_snapshot_display.py
+6 -5
View File
@@ -11,6 +11,7 @@ from PIL import Image, UnidentifiedImageError
import pytest
from .bmp_utils import Bmp, capture_when_drawn, wait_for_bmp
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
WIDTH = 101
@@ -81,12 +82,12 @@ async def test_snapshot_display(
async with run_compiled(yaml_config), api_client_connected() as client:
_, services = await client.list_entities_services()
service = next(s for s in services if s.name == "take_snapshot")
service = require_service(services, "take_snapshot")
async def take(name: str) -> None:
await client.execute_service(service, {"name": name})
animation_service = next(s for s in services if s.name == "take_animation")
animation_service = require_service(services, "take_animation")
# The test card is drawn in several colours, so once it is on the screen the picture is
# not one flat shade. Capturing until that is true waits out the first update rather than
@@ -110,8 +111,8 @@ async def test_snapshot_display(
# it passes the point where the code width grows and where the dictionary is started over.
# The picture uses no more than 256 colours, so it is stored exactly and every frame must
# come back identical to the one captured as a BMP.
noise_service = next(s for s in services if s.name == "take_noise_snapshot")
noise_animation = next(s for s in services if s.name == "take_noise_animation")
noise_service = require_service(services, "take_noise_snapshot")
noise_animation = require_service(services, "take_noise_animation")
async def take_noise(name: str) -> None:
await client.execute_service(noise_service, {"name": name})
@@ -155,7 +156,7 @@ async def test_snapshot_display(
# A display records one animation at a time. A second request made while the first is still
# running is refused and leaves no file, and the first one carries on to the end.
slow_animation = next(s for s in services if s.name == "take_slow_animation")
slow_animation = require_service(services, "take_slow_animation")
await client.execute_service(slow_animation, {"name": "slow"})
await client.execute_service(slow_animation, {"name": "second"})
await asyncio.wait_for(busy.wait(), timeout=10.0)
tests/integration/test_socket_wake_gate_tcp.py
+2 -3
View File
@@ -17,7 +17,7 @@ import time
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -51,8 +51,7 @@ async def test_socket_wake_gate_tcp(
)
_, services = await client.list_entities_services()
ping_service = next((s for s in services if s.name == "ping"), None)
assert ping_service is not None, "ping service not found"
ping_service = require_service(services, "ping")
# Execute the service and time how long until PONG appears in logs.
# The request bytes arrive on an accept_loop_monitored() TCP socket,
tests/integration/test_syslog.py
+3 -9
View File
@@ -13,7 +13,7 @@ from typing import TypedDict
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -252,10 +252,7 @@ async def test_syslog(
# Test message truncation - call service that logs a very long message
_, services = await client.list_entities_services()
log_service = next(
(s for s in services if s.name == "log_long_message"), None
)
assert log_service is not None, "log_long_message service not found"
log_service = require_service(services, "log_long_message")
# Call the service to trigger a long log message
await client.execute_service(log_service, {})
@@ -278,10 +275,7 @@ async def test_syslog(
)
# Test short message - should arrive complete (not truncated)
short_service = next(
(s for s in services if s.name == "log_short_message"), None
)
assert short_service is not None, "log_short_message service not found"
short_service = require_service(services, "log_short_message")
await client.execute_service(short_service, {})
tests/integration/test_tcp_uart_bridge.py
+2 -2
View File
@@ -13,7 +13,7 @@ import contextlib
import pytest
from .log_utils import LineWaiter
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -74,7 +74,7 @@ async def test_tcp_uart_bridge(
):
await assert_device_name(client, "tcp-uart-bridge-test")
_, services = await client.list_entities_services()
send_byte = next(s for s in services if s.name == "send_byte")
send_byte = require_service(services, "send_byte")
await wait_or_fail(
echo_ok, 15.0, "UART echo through the bridge never arrived"
tests/integration/test_udp.py
+3 -11
View File
@@ -11,7 +11,7 @@ import socket
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -135,12 +135,7 @@ async def test_udp_send_receive(
_, services = await client.list_entities_services()
# Test sending string message
send_message_service = next(
(s for s in services if s.name == "send_udp_message"), None
)
assert send_message_service is not None, (
"send_udp_message service not found"
)
send_message_service = require_service(services, "send_udp_message")
await client.execute_service(send_message_service, {})
@@ -153,10 +148,7 @@ async def test_udp_send_receive(
)
# Test sending bytes
send_bytes_service = next(
(s for s in services if s.name == "send_udp_bytes"), None
)
assert send_bytes_service is not None, "send_udp_bytes service not found"
send_bytes_service = require_service(services, "send_udp_bytes")
await client.execute_service(send_bytes_service, {})
tests/integration/test_wait_until_mid_loop_timing.py
+2 -4
View File
@@ -16,6 +16,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -80,10 +81,7 @@ async def test_wait_until_mid_loop_timing(
):
# Get the test service
_, services = await client.list_entities_services()
test_service = next(
(s for s in services if s.name == "test_mid_loop_timeout"), None
)
assert test_service is not None, "test_mid_loop_timeout service not found"
test_service = require_service(services, "test_mid_loop_timeout")
# Execute the test
await client.execute_service(test_service, {})
tests/integration/test_wait_until_on_boot.py
+2 -4
View File
@@ -17,6 +17,7 @@ import re
import pytest
from .state_utils import require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
@@ -69,10 +70,7 @@ async def test_wait_until_on_boot(
# Now set the flag that wait_until is waiting for
_, services = await client.list_entities_services()
set_flag_service = next(
(s for s in services if s.name == "set_test_flag"), None
)
assert set_flag_service is not None, "set_test_flag service not found"
set_flag_service = require_service(services, "set_test_flag")
await client.execute_service(set_flag_service, {})
tests/integration/test_wait_until_ordering.py
+2 -5
View File
@@ -14,7 +14,7 @@ import re
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -65,10 +65,7 @@ async def test_wait_until_fifo_ordering(
# Get services
_, services = await client.list_entities_services()
test_service = next(
(s for s in services if s.name == "test_wait_until_fifo"), None
)
assert test_service is not None, "test_wait_until_fifo service not found"
test_service = require_service(services, "test_wait_until_fifo")
# Execute the test
await client.execute_service(test_service, {})
tests/integration/test_wait_until_reentrant_restart.py
+3 -9
View File
@@ -15,7 +15,7 @@ import re
import pytest
from .state_utils import assert_device_name
from .state_utils import assert_device_name, require_service
from .types import APIClientConnectedFactory, RunCompiledFunction
from .wait_utils import wait_or_fail
@@ -54,14 +54,8 @@ async def test_wait_until_reentrant_restart(
await assert_device_name(client, "wait-until-reentrant-restart")
_, services = await client.list_entities_services()
self_restart_service = next(
(s for s in services if s.name == "start_self_restart"), None
)
assert self_restart_service is not None, "start_self_restart not found"
stop_service = next(
(s for s in services if s.name == "start_stop_during_wait"), None
)
assert stop_service is not None, "start_stop_during_wait not found"
self_restart_service = require_service(services, "start_self_restart")
stop_service = require_service(services, "start_stop_during_wait")
# Scenario 1: the wait_until timeout continuation restarts its own
# script five times, re-entering the same wait_until each time.