VoIP test improvements (#179456)

tests/components/voip/conftest.py

@@ -1,17 +1,22 @@
 """Test helpers for VoIP integration."""
 
 from collections.abc import Generator
-from unittest.mock import AsyncMock, Mock, patch
+from unittest.mock import AsyncMock, Mock, create_autospec, patch
 
 import pytest
 from voip_utils import CallInfo
 from voip_utils.sip import get_sip_endpoint
 
+from homeassistant.components import assist_satellite, voip
+from homeassistant.components.assist_satellite import AssistSatelliteEntity
 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.const import Platform
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import entity_registry as er
+from homeassistant.helpers.entity_component import EntityComponent
 from homeassistant.setup import async_setup_component
 
 from tests.common import MockConfigEntry
@@ -125,3 +130,39 @@ async def voip_device(
     # to make sure all platforms are set up
     await hass.async_block_till_done()
     return device
+
+
+@pytest.fixture
+def satellite(
+    hass: HomeAssistant,
+    voip_device: VoIPDevice,
+):
+    """Create VoipAssistSatellite for use in tests."""
+    satellite = async_get_satellite_entity(hass, voip.DOMAIN, voip_device.voip_id)
+    assert isinstance(satellite, VoipAssistSatellite)
+
+    mock_send_audio = create_autospec(satellite.send_audio)
+    satellite.send_audio = mock_send_audio
+
+    yield satellite
+
+    if satellite.voip_device.is_active:
+        satellite.disconnect()
+
+
+def async_get_satellite_entity(
+    hass: HomeAssistant, domain: str, unique_id_prefix: str
+) -> AssistSatelliteEntity | None:
+    """Get Assist satellite entity."""
+    ent_reg = er.async_get(hass)
+    satellite_entity_id = ent_reg.async_get_entity_id(
+        Platform.ASSIST_SATELLITE, domain, f"{unique_id_prefix}-assist_satellite"
+    )
+    if satellite_entity_id is None:
+        return None
+    assert not satellite_entity_id.endswith("none")
+
+    component: EntityComponent[AssistSatelliteEntity] = hass.data[
+        assist_satellite.DOMAIN
+    ]
+    return component.get_entity(satellite_entity_id)

tests/components/voip/test_voip.py

@@ -12,7 +12,6 @@ from syrupy.assertion import SnapshotAssertion
 from voip_utils import CallInfo
 
 from homeassistant.components import assist_pipeline, assist_satellite, tts, voip
-from homeassistant.components.assist_satellite import AssistSatelliteEntity
 
 # pylint: disable-next=home-assistant-component-root-import
 from homeassistant.components.assist_satellite.entity import AssistSatelliteState
@@ -20,11 +19,10 @@ from homeassistant.components.voip import DOMAIN, HassVoipDatagramProtocol
 from homeassistant.components.voip.assist_satellite import Tones, VoipAssistSatellite
 from homeassistant.components.voip.devices import VoIPDevice, VoIPDevices
 from homeassistant.components.voip.voip import PreRecordMessageProtocol, make_protocol
-from homeassistant.const import STATE_OFF, STATE_ON, Platform
+from homeassistant.const import STATE_OFF, STATE_ON
 from homeassistant.core import Context, HomeAssistant
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import entity_registry as er
-from homeassistant.helpers.entity_component import EntityComponent
 from homeassistant.setup import async_setup_component
 
 from tests.common import MockConfigEntry
@@ -39,21 +37,6 @@ def mock_tts_cache_dir_autouse(mock_tts_cache_dir: Path) -> None:
     """Mock the TTS cache dir with empty dir."""
 
 
-@pytest.fixture
-def satellite(
-    hass: HomeAssistant,
-    voip_device: VoIPDevice,
-):
-    """Create VoipAssistSatellite for use in tests."""
-    satellite = async_get_satellite_entity(hass, voip.DOMAIN, voip_device.voip_id)
-    assert isinstance(satellite, VoipAssistSatellite)
-
-    yield satellite
-
-    if satellite.voip_device.is_active:
-        satellite.disconnect()
-
-
 def _empty_wav(framerate=16000) -> bytes:
     """Return bytes of an empty WAV file."""
     with io.BytesIO() as wav_io:
@@ -66,24 +49,6 @@ def _empty_wav(framerate=16000) -> bytes:
         return wav_io.getvalue()
 
 
-def async_get_satellite_entity(
-    hass: HomeAssistant, domain: str, unique_id_prefix: str
-) -> AssistSatelliteEntity | None:
-    """Get Assist satellite entity."""
-    ent_reg = er.async_get(hass)
-    satellite_entity_id = ent_reg.async_get_entity_id(
-        Platform.ASSIST_SATELLITE, domain, f"{unique_id_prefix}-assist_satellite"
-    )
-    if satellite_entity_id is None:
-        return None
-    assert not satellite_entity_id.endswith("none")
-
-    component: EntityComponent[AssistSatelliteEntity] = hass.data[
-        assist_satellite.DOMAIN
-    ]
-    return component.get_entity(satellite_entity_id)
-
-
 async def test_is_valid_call(
     hass: HomeAssistant,
     voip_devices: VoIPDevices,
@@ -483,7 +448,6 @@ async def test_tts_timeout(
             satellite._tone_bytes[tone] = tone_bytes
 
         satellite.connection_made(Mock())
-        satellite.send_audio = Mock()
 
         original_send_tts = satellite._send_tts
 
@@ -588,14 +552,7 @@ async def test_tts_wrong_extension(
 
         # silence (assumes relaxed VAD sensitivity)
         satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
+        await asyncio.sleep(0.2)
 
         # Wait for mock pipeline to exhaust the audio stream
         async with asyncio.timeout(3):
@@ -679,14 +636,7 @@ async def test_tts_wrong_wav_format(
 
         # silence (assumes relaxed VAD sensitivity)
         satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
-        await asyncio.sleep(0.05)
-        satellite.on_chunk(bytes(_ONE_SECOND))
+        await asyncio.sleep(0.2)
 
         # Wait for mock pipeline to exhaust the audio stream
         async with asyncio.timeout(3):