starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Speed up VoIP tests (#179786)

Browse Source 
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Paulus Schoutsen
2026-08-22 13:52:42 +02:00
committed by GitHub
co-authored by Claude
parent 41d1fe77f3
commit a3bb68c87e
2 changed files with 59 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/components/voip/conftest.py
+36
View File
@@ -1,5 +1,6 @@
"""Test helpers for VoIP integration."""
from collections.abc import Generator
from unittest.mock import AsyncMock, Mock, patch
import pytest
@@ -7,6 +8,7 @@ from voip_utils import CallInfo
from voip_utils.sip import get_sip_endpoint
from homeassistant.components.voip import DOMAIN
from homeassistant.components.voip.assist_satellite import VoipAssistSatellite
from homeassistant.components.voip.devices import VoIPDevice, VoIPDevices
from homeassistant.config_entries import ConfigEntryState
from homeassistant.core import HomeAssistant
@@ -24,6 +26,40 @@ async def load_homeassistant(hass: HomeAssistant) -> None:
assert await async_setup_component(hass, "homeassistant", {})
@pytest.fixture(autouse=True)
def reduce_satellite_delays() -> Generator[None]:
"""Shorten the delays that the satellite always waits out.
Tests must send audio chunks more often than _HANGUP_SEC, or the satellite
treats the gap as the caller hanging up. Timeouts that only elapse when audio
never arrives are left alone: they cost nothing unless a test exercises them.
"""
with (
patch("homeassistant.components.voip.assist_satellite._HANGUP_SEC", 0.2),
patch(
"homeassistant.components.voip.assist_satellite._ANNOUNCEMENT_BEFORE_DELAY",
0.1,
),
patch(
"homeassistant.components.voip.assist_satellite._ANNOUNCEMENT_AFTER_DELAY",
0.1,
),
):
yield
@pytest.fixture
def silent_tones() -> Generator[None]:
"""Give every tone empty audio.
A real tone is up to two seconds streamed in real time, which outlasts the
shortened hangup window. The tone paths still run, so the processing tone
still gates _send_tts.
"""
with patch.object(VoipAssistSatellite, "_load_pcm", return_value=b""):
yield
@pytest.fixture
def config_entry(hass: HomeAssistant) -> MockConfigEntry:
"""Create a config entry."""
tests/components/voip/test_voip.py
+23 -28
View File
@@ -511,6 +511,7 @@ async def test_tts_timeout(
await done.wait()
@pytest.mark.usefixtures("silent_tones")
async def test_tts_wrong_extension(
hass: HomeAssistant,
satellite: VoipAssistSatellite,
@@ -587,13 +588,13 @@ async def test_tts_wrong_extension(
# silence (assumes relaxed VAD sensitivity)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
# Wait for mock pipeline to exhaust the audio stream
@@ -601,6 +602,7 @@ async def test_tts_wrong_extension(
await done.wait()
@pytest.mark.usefixtures("silent_tones")
async def test_tts_wrong_wav_format(
hass: HomeAssistant,
satellite: VoipAssistSatellite,
@@ -677,13 +679,13 @@ async def test_tts_wrong_wav_format(
# silence (assumes relaxed VAD sensitivity)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
# Wait for mock pipeline to exhaust the audio stream
@@ -691,6 +693,7 @@ async def test_tts_wrong_wav_format(
await done.wait()
@pytest.mark.usefixtures("silent_tones")
async def test_empty_tts_output(
hass: HomeAssistant,
satellite: VoipAssistSatellite,
@@ -757,16 +760,9 @@ async def test_empty_tts_output(
# silence (assumes relaxed VAD sensitivity)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
satellite.on_chunk(bytes(_ONE_SECOND))
# Wait for mock pipeline to finish
# No more chunks: another chunk would start a second pipeline run, which
# clears _tts_done again.
async with asyncio.timeout(2):
await satellite._tts_done.wait()
@@ -877,9 +873,9 @@ async def test_announce(
# Trigger announcement
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
async with asyncio.timeout(2):
await announce_task
@@ -937,9 +933,9 @@ async def test_voip_id_is_ip_address(
# Trigger announcement
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
async with asyncio.timeout(2):
await announce_task
@@ -1032,11 +1028,11 @@ async def test_announce_disconnect(
# Trigger announcement
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
assert satellite._announcement is announcement
assert voip_device.is_active
@@ -1196,18 +1192,17 @@ async def test_start_conversation(
# Trigger announcement and wait for it to finish
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
async with asyncio.timeout(2):
await tts_sent.wait()
# Trigger pipeline
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(0.2)
await asyncio.sleep(0.05)
satellite.on_chunk(bytes(_ONE_SECOND))
await asyncio.sleep(3)
async with asyncio.timeout(3):
# Wait for Conversation end
await conversation_task