mirror of https://github.com/home-assistant/core
474 lines
15 KiB
Python
474 lines
15 KiB
Python
"""Test Home Assistant timeout handler."""
|
|
|
|
import asyncio
|
|
from contextlib import suppress
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.util.timeout import TimeoutManager
|
|
|
|
|
|
async def test_simple_global_timeout() -> None:
|
|
"""Test a simple global timeout."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError):
|
|
async with timeout.async_timeout(0.1):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_global_timeout_with_executor_job(hass: HomeAssistant) -> None:
|
|
"""Test a simple global timeout with executor job."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError):
|
|
async with timeout.async_timeout(0.1):
|
|
await hass.async_add_executor_job(time.sleep, 0.2)
|
|
|
|
|
|
async def test_simple_global_timeout_freeze() -> None:
|
|
"""Test a simple global timeout freeze."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with timeout.async_timeout(0.2), timeout.async_freeze():
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_zone_timeout_freeze_inside_executor_job(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple zone timeout freeze inside an executor job."""
|
|
timeout = TimeoutManager()
|
|
|
|
def _some_sync_work():
|
|
with timeout.freeze("recorder"):
|
|
time.sleep(0.3)
|
|
|
|
async with (
|
|
timeout.async_timeout(1.0),
|
|
timeout.async_timeout(0.2, zone_name="recorder"),
|
|
):
|
|
await hass.async_add_executor_job(_some_sync_work)
|
|
|
|
|
|
async def test_simple_global_timeout_freeze_inside_executor_job(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple global timeout freeze inside an executor job."""
|
|
timeout = TimeoutManager()
|
|
|
|
def _some_sync_work():
|
|
with timeout.freeze():
|
|
time.sleep(0.3)
|
|
|
|
async with timeout.async_timeout(0.2):
|
|
await hass.async_add_executor_job(_some_sync_work)
|
|
|
|
|
|
async def test_mix_global_timeout_freeze_and_zone_freeze_inside_executor_job(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple global timeout freeze inside an executor job."""
|
|
timeout = TimeoutManager()
|
|
|
|
def _some_sync_work():
|
|
with timeout.freeze("recorder"):
|
|
time.sleep(0.3)
|
|
|
|
async with (
|
|
timeout.async_timeout(0.1),
|
|
timeout.async_timeout(0.2, zone_name="recorder"),
|
|
):
|
|
await hass.async_add_executor_job(_some_sync_work)
|
|
|
|
|
|
async def test_mix_global_timeout_freeze_and_zone_freeze_different_order(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple global timeout freeze inside an executor job before timeout was set."""
|
|
timeout = TimeoutManager()
|
|
|
|
def _some_sync_work():
|
|
with timeout.freeze("recorder"):
|
|
time.sleep(0.4)
|
|
|
|
async with timeout.async_timeout(0.1):
|
|
hass.async_add_executor_job(_some_sync_work)
|
|
async with timeout.async_timeout(0.2, zone_name="recorder"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_mix_global_timeout_freeze_and_zone_freeze_other_zone_inside_executor_job(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple global timeout freeze other zone inside an executor job."""
|
|
timeout = TimeoutManager()
|
|
|
|
def _some_sync_work():
|
|
with timeout.freeze("not_recorder"):
|
|
time.sleep(0.3)
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1):
|
|
async with (
|
|
timeout.async_timeout(0.2, zone_name="recorder"),
|
|
timeout.async_timeout(0.2, zone_name="not_recorder"),
|
|
):
|
|
await hass.async_add_executor_job(_some_sync_work)
|
|
|
|
|
|
async def test_mix_global_timeout_freeze_and_zone_freeze_inside_executor_job_second_job_outside_zone_context(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple global timeout freeze inside an executor job with second job outside of zone context."""
|
|
timeout = TimeoutManager()
|
|
|
|
def _some_sync_work():
|
|
with timeout.freeze("recorder"):
|
|
time.sleep(0.3)
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1):
|
|
async with timeout.async_timeout(0.2, zone_name="recorder"):
|
|
await hass.async_add_executor_job(_some_sync_work)
|
|
await hass.async_add_executor_job(time.sleep, 0.2)
|
|
|
|
|
|
async def test_simple_global_timeout_freeze_with_executor_job(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple global timeout freeze with executor job."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with timeout.async_timeout(0.2), timeout.async_freeze():
|
|
await hass.async_add_executor_job(time.sleep, 0.3)
|
|
|
|
|
|
async def test_simple_global_timeout_does_not_leak_upward(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a global timeout does not leak upward."""
|
|
timeout = TimeoutManager()
|
|
current_task = asyncio.current_task()
|
|
assert current_task is not None
|
|
cancelling_inside_timeout = None
|
|
|
|
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1):
|
|
cancelling_inside_timeout = current_task.cancelling()
|
|
await asyncio.sleep(0.3)
|
|
|
|
assert cancelling_inside_timeout == 0
|
|
# After the context manager exits, the task should no longer be cancelling
|
|
assert current_task.cancelling() == 0
|
|
|
|
|
|
async def test_simple_global_timeout_does_swallow_cancellation(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a global timeout does not swallow cancellation."""
|
|
timeout = TimeoutManager()
|
|
current_task = asyncio.current_task()
|
|
assert current_task is not None
|
|
cancelling_inside_timeout = None
|
|
|
|
async def task_with_timeout() -> None:
|
|
nonlocal cancelling_inside_timeout
|
|
new_task = asyncio.current_task()
|
|
assert new_task is not None
|
|
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
|
|
cancelling_inside_timeout = new_task.cancelling()
|
|
async with timeout.async_timeout(0.1):
|
|
await asyncio.sleep(0.3)
|
|
|
|
# After the context manager exits, the task should no longer be cancelling
|
|
assert current_task.cancelling() == 0
|
|
|
|
task = asyncio.create_task(task_with_timeout())
|
|
await asyncio.sleep(0)
|
|
task.cancel()
|
|
assert task.cancelling() == 1
|
|
|
|
assert cancelling_inside_timeout == 0
|
|
# Cancellation should not leak into the current task
|
|
assert current_task.cancelling() == 0
|
|
# Cancellation should not be swallowed if the task is cancelled
|
|
# and it also times out
|
|
await asyncio.sleep(0)
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert task.cancelling() == 1
|
|
|
|
|
|
async def test_simple_global_timeout_freeze_reset() -> None:
|
|
"""Test a simple global timeout freeze reset."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.2):
|
|
async with timeout.async_freeze():
|
|
await asyncio.sleep(0.1)
|
|
await asyncio.sleep(0.2)
|
|
|
|
|
|
async def test_simple_zone_timeout() -> None:
|
|
"""Test a simple zone timeout."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError):
|
|
async with timeout.async_timeout(0.1, "test"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_zone_timeout_does_not_leak_upward(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a zone timeout does not leak upward."""
|
|
timeout = TimeoutManager()
|
|
current_task = asyncio.current_task()
|
|
assert current_task is not None
|
|
cancelling_inside_timeout = None
|
|
|
|
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1, "test"):
|
|
cancelling_inside_timeout = current_task.cancelling()
|
|
await asyncio.sleep(0.3)
|
|
|
|
assert cancelling_inside_timeout == 0
|
|
# After the context manager exits, the task should no longer be cancelling
|
|
assert current_task.cancelling() == 0
|
|
|
|
|
|
async def test_simple_zone_timeout_does_swallow_cancellation(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a zone timeout does not swallow cancellation."""
|
|
timeout = TimeoutManager()
|
|
current_task = asyncio.current_task()
|
|
assert current_task is not None
|
|
cancelling_inside_timeout = None
|
|
|
|
async def task_with_timeout() -> None:
|
|
nonlocal cancelling_inside_timeout
|
|
new_task = asyncio.current_task()
|
|
assert new_task is not None
|
|
with pytest.raises(asyncio.TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1, "test"):
|
|
cancelling_inside_timeout = current_task.cancelling()
|
|
await asyncio.sleep(0.3)
|
|
|
|
# After the context manager exits, the task should no longer be cancelling
|
|
assert current_task.cancelling() == 0
|
|
|
|
task = asyncio.create_task(task_with_timeout())
|
|
await asyncio.sleep(0)
|
|
task.cancel()
|
|
assert task.cancelling() == 1
|
|
|
|
# Cancellation should not leak into the current task
|
|
assert cancelling_inside_timeout == 0
|
|
assert current_task.cancelling() == 0
|
|
# Cancellation should not be swallowed if the task is cancelled
|
|
# and it also times out
|
|
await asyncio.sleep(0)
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert task.cancelling() == 1
|
|
|
|
|
|
async def test_multiple_zone_timeout() -> None:
|
|
"""Test a simple zone timeout."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1, "test"):
|
|
async with timeout.async_timeout(0.5, "test"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_different_zone_timeout() -> None:
|
|
"""Test a simple zone timeout."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1, "test"):
|
|
async with timeout.async_timeout(0.5, "other"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_zone_timeout_freeze() -> None:
|
|
"""Test a simple zone timeout freeze."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with timeout.async_timeout(0.2, "test"), timeout.async_freeze("test"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_zone_timeout_freeze_without_timeout() -> None:
|
|
"""Test a simple zone timeout freeze on a zone that does not have a timeout set."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with timeout.async_timeout(0.1, "test"), timeout.async_freeze("test"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_zone_timeout_freeze_reset() -> None:
|
|
"""Test a simple zone timeout freeze reset."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.2, "test"):
|
|
async with timeout.async_freeze("test"):
|
|
await asyncio.sleep(0.1)
|
|
await asyncio.sleep(0.2, "test")
|
|
|
|
|
|
async def test_mix_zone_timeout_freeze_and_global_freeze() -> None:
|
|
"""Test a mix zone timeout freeze and global freeze."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with (
|
|
timeout.async_timeout(0.2, "test"),
|
|
timeout.async_freeze("test"),
|
|
timeout.async_freeze(),
|
|
):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_mix_global_and_zone_timeout_freeze_() -> None:
|
|
"""Test a mix zone timeout freeze and global freeze."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with (
|
|
timeout.async_timeout(0.2, "test"),
|
|
timeout.async_freeze(),
|
|
timeout.async_freeze("test"),
|
|
):
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_mix_zone_timeout_freeze() -> None:
|
|
"""Test a mix zone timeout global freeze."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with timeout.async_timeout(0.2, "test"), timeout.async_freeze():
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_mix_zone_timeout() -> None:
|
|
"""Test a mix zone timeout global."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with timeout.async_timeout(0.1):
|
|
with suppress(TimeoutError):
|
|
async with timeout.async_timeout(0.2, "test"):
|
|
await asyncio.sleep(0.4)
|
|
|
|
|
|
async def test_mix_zone_timeout_trigger_global() -> None:
|
|
"""Test a mix zone timeout global with trigger it."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1):
|
|
with suppress(TimeoutError):
|
|
async with timeout.async_timeout(0.1, "test"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_mix_zone_timeout_trigger_global_cool_down() -> None:
|
|
"""Test a mix zone timeout global with trigger it with cool_down."""
|
|
timeout = TimeoutManager()
|
|
|
|
async with timeout.async_timeout(0.1, cool_down=0.3):
|
|
with suppress(TimeoutError):
|
|
async with timeout.async_timeout(0.1, "test"):
|
|
await asyncio.sleep(0.3)
|
|
|
|
await asyncio.sleep(0.2)
|
|
|
|
# Cleanup lingering (cool_down) task after test is done
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_zone_timeout_freeze_without_timeout_cleanup(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple zone timeout freeze on a zone that does not have a timeout set."""
|
|
timeout = TimeoutManager()
|
|
|
|
async def background():
|
|
async with timeout.async_freeze("test"):
|
|
await asyncio.sleep(0.4)
|
|
|
|
async with timeout.async_timeout(0.1):
|
|
hass.async_create_task(background())
|
|
await asyncio.sleep(0.2)
|
|
|
|
|
|
async def test_simple_zone_timeout_freeze_without_timeout_cleanup2(
|
|
hass: HomeAssistant,
|
|
) -> None:
|
|
"""Test a simple zone timeout freeze on a zone that does not have a timeout set."""
|
|
timeout = TimeoutManager()
|
|
|
|
async def background():
|
|
async with timeout.async_freeze("test"):
|
|
await asyncio.sleep(0.2)
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1):
|
|
hass.async_create_task(background())
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_simple_zone_timeout_freeze_without_timeout_exeption() -> None:
|
|
"""Test a simple zone timeout freeze on a zone that does not have a timeout set."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1):
|
|
with suppress(RuntimeError):
|
|
async with timeout.async_freeze("test"):
|
|
raise RuntimeError
|
|
|
|
await asyncio.sleep(0.4)
|
|
|
|
|
|
async def test_simple_zone_timeout_zone_with_timeout_exception() -> None:
|
|
"""Test a simple zone timeout freeze on a zone that does not have a timeout set."""
|
|
timeout = TimeoutManager()
|
|
|
|
with pytest.raises(TimeoutError): # noqa: PT012
|
|
async with timeout.async_timeout(0.1):
|
|
with suppress(RuntimeError):
|
|
async with timeout.async_timeout(0.3, "test"):
|
|
raise RuntimeError
|
|
|
|
await asyncio.sleep(0.3)
|
|
|
|
|
|
async def test_multiple_global_freezes(hass: HomeAssistant) -> None:
|
|
"""Test multiple global freezes."""
|
|
timeout = TimeoutManager()
|
|
|
|
async def background(delay: float) -> None:
|
|
async with timeout.async_freeze():
|
|
await asyncio.sleep(delay)
|
|
|
|
async with timeout.async_timeout(0.1):
|
|
task = hass.async_create_task(background(0.2))
|
|
async with timeout.async_freeze():
|
|
await asyncio.sleep(0.1)
|
|
await task
|
|
|
|
async with timeout.async_timeout(0.1):
|
|
task = hass.async_create_task(background(0.2))
|
|
async with timeout.async_freeze():
|
|
await asyncio.sleep(0.3)
|
|
await task
|