Home-Assistant/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-05 04:51:05 +01:00
Code Activity

Use Home Assistant's shared aiohttp session in Lyngdorf (#180620)

Browse Source
This commit is contained in:
Alex Fishlock
2026-08-29 15:25:10 +02:00
committed by GitHub
parent 6e287f4b23
commit 8094598c2f
6 changed files with 116 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/lyngdorf/__init__.py
+4 -1
View File
@@ -8,6 +8,7 @@ from homeassistant.const import CONF_HOST, CONF_MODEL, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import (
CONNECTION_NETWORK_MAC,
DeviceInfo,
@@ -42,7 +43,9 @@ async def async_setup_entry(
try:
receiver: LyngdorfReceiver = await create_receiver(
config_entry.data[CONF_HOST], lyngdorf_model
config_entry.data[CONF_HOST],
lyngdorf_model,
session=async_get_clientsession(hass),
)
await receiver.connect()
except TimeoutError as err:
homeassistant/components/lyngdorf/config_flow.py
+8 -1
View File
@@ -17,6 +17,7 @@ from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_MODEL
from homeassistant.data_entry_flow import AbortFlow
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MODEL_NAME,
@@ -100,7 +101,13 @@ class LyngdorfFlowHandler(ConfigFlow, domain=DOMAIN):
try:
location = await discover_ssdp_location(host)
serial = await fetch_device_serial(location) if location else None
serial = (
await fetch_device_serial(
location, session=async_get_clientsession(self.hass)
)
if location
else None
)
except TimeoutError as err:
raise TimeoutConnect from err
except OSError as err:
homeassistant/components/lyngdorf/quality_scale.yaml
+1 -3
View File
@@ -84,7 +84,5 @@ rules:
# Platinum
async-dependency: todo
inject-websession:
status: exempt
comment: Integration uses local TCP, not HTTP.
inject-websession: done
strict-typing: todo
tests/components/lyngdorf/conftest.py
+83 -77
View File
@@ -55,92 +55,98 @@ def mock_setup_entry() -> Generator[None]:
@pytest.fixture
def mock_receiver() -> Generator[MagicMock]:
"""Return a mocked Lyngdorf receiver."""
def mock_create_receiver() -> Generator[MagicMock]:
"""Return a mocked create_receiver factory."""
with patch("homeassistant.components.lyngdorf.create_receiver") as create_mock:
receiver = MagicMock(spec=LyngdorfReceiver)
receiver.name = "Mock Lyngdorf"
receiver.connected = True
receiver.has_remote_keys = True
receiver.available_remote_keys = frozenset(
{
RemoteKey.UP,
RemoteKey.DOWN,
RemoteKey.ENTER,
RemoteKey.MENU,
RemoteKey.DIGIT_0,
}
)
yield create_mock
# Diagnostics reports the whole receiver, so every property it reads
# needs a value here; an unset one is a mock the response cannot encode.
receiver.model = LyngdorfModel.MP_60
receiver.max_volume = 0.0
receiver.room_perfect_position = "Focus 1"
receiver.available_room_perfect_positions = ["Global", "Focus 1"]
receiver.voicing = "Neutral"
receiver.available_voicings = ["Neutral", "Music", "Movie"]
receiver.lipsync = None
receiver.lipsync_range = NumericRange(0, 500, 1)
for _t in ("bass", "treble"):
setattr(receiver, f"trim_{_t}", None)
setattr(receiver, f"trim_{_t}_range", NumericRange(-12.0, 12.0, 0.1))
for _t in ("centre", "height", "lfe", "surround"):
setattr(receiver, f"trim_{_t}", None)
setattr(receiver, f"trim_{_t}_range", NumericRange(-10.0, 10.0, 0.1))
receiver.volume_range = NumericRange(-99.9, 24.0, 0.1)
receiver.zone_b_volume_range = NumericRange(-99.9, 24.0, 0.1)
@pytest.fixture
def mock_receiver(mock_create_receiver: MagicMock) -> MagicMock:
"""Return a mocked Lyngdorf receiver."""
receiver = MagicMock(spec=LyngdorfReceiver)
receiver.name = "Mock Lyngdorf"
receiver.connected = True
receiver.has_remote_keys = True
receiver.available_remote_keys = frozenset(
{
RemoteKey.UP,
RemoteKey.DOWN,
RemoteKey.ENTER,
RemoteKey.MENU,
RemoteKey.DIGIT_0,
}
)
receiver.power_on = False
receiver.volume = -40.0
receiver.mute_enabled = False
receiver.source = None
receiver.available_sources = []
receiver.sound_mode = None
receiver.available_sound_modes = []
# Diagnostics reports the whole receiver, so every property it reads
# needs a value here; an unset one is a mock the response cannot encode.
receiver.model = LyngdorfModel.MP_60
receiver.max_volume = 0.0
receiver.room_perfect_position = "Focus 1"
receiver.available_room_perfect_positions = ["Global", "Focus 1"]
receiver.voicing = "Neutral"
receiver.available_voicings = ["Neutral", "Music", "Movie"]
receiver.lipsync = None
receiver.lipsync_range = NumericRange(0, 500, 1)
for _t in ("bass", "treble"):
setattr(receiver, f"trim_{_t}", None)
setattr(receiver, f"trim_{_t}_range", NumericRange(-12.0, 12.0, 0.1))
for _t in ("centre", "height", "lfe", "surround"):
setattr(receiver, f"trim_{_t}", None)
setattr(receiver, f"trim_{_t}_range", NumericRange(-10.0, 10.0, 0.1))
receiver.audio_information = "Stereo"
receiver.video_information = "4K HDR"
receiver.audio_input = "optical"
receiver.video_input = "hdmi"
receiver.streaming_source = "AirPlay"
receiver.available_audio_inputs = ["optical", "aux"]
receiver.available_video_inputs = ["hdmi"]
receiver.available_stream_types = ["AirPlay", "DLNA"]
receiver.volume_range = NumericRange(-99.9, 24.0, 0.1)
receiver.zone_b_volume_range = NumericRange(-99.9, 24.0, 0.1)
receiver.now_playing = None
receiver.has_position = False
receiver.position_ms = None
receiver.position_updated_at = None
receiver.shuffle = None
receiver.repeat = None
receiver.can_shuffle = False
receiver.available_repeat_modes = frozenset()
receiver.power_on = False
receiver.volume = -40.0
receiver.mute_enabled = False
receiver.source = None
receiver.available_sources = []
receiver.sound_mode = None
receiver.available_sound_modes = []
receiver.lipsync = 50
receiver.lipsync_range = NumericRange(0, 500, 1)
receiver.trim_bass = 3.0
receiver.trim_treble = 0.0
receiver.trim_centre = 0.0
receiver.trim_height = 4.0
receiver.trim_lfe = 3.0
receiver.trim_surround = 0.0
receiver.trim_bass_range = NumericRange(-12.0, 12.0, 0.1)
receiver.trim_treble_range = NumericRange(-12.0, 12.0, 0.1)
for _trim in ("centre", "height", "lfe", "surround"):
setattr(receiver, f"trim_{_trim}_range", NumericRange(-10.0, 10.0, 0.1))
receiver.audio_information = "Stereo"
receiver.video_information = "4K HDR"
receiver.audio_input = "optical"
receiver.video_input = "hdmi"
receiver.streaming_source = "AirPlay"
receiver.available_audio_inputs = ["optical", "aux"]
receiver.available_video_inputs = ["hdmi"]
receiver.available_stream_types = ["AirPlay", "DLNA"]
receiver.zone_b_power_on = False
receiver.zone_b_volume = -40.0
receiver.zone_b_mute_enabled = False
receiver.zone_b_source = None
receiver.zone_b_available_sources = []
receiver.zone_b_audio_input = "aux"
receiver.zone_b_streaming_source = "DLNA"
receiver.now_playing = None
receiver.has_position = False
receiver.position_ms = None
receiver.position_updated_at = None
receiver.shuffle = None
receiver.repeat = None
receiver.can_shuffle = False
receiver.available_repeat_modes = frozenset()
create_mock.return_value = receiver
yield receiver
receiver.lipsync = 50
receiver.lipsync_range = NumericRange(0, 500, 1)
receiver.trim_bass = 3.0
receiver.trim_treble = 0.0
receiver.trim_centre = 0.0
receiver.trim_height = 4.0
receiver.trim_lfe = 3.0
receiver.trim_surround = 0.0
receiver.trim_bass_range = NumericRange(-12.0, 12.0, 0.1)
receiver.trim_treble_range = NumericRange(-12.0, 12.0, 0.1)
for _trim in ("centre", "height", "lfe", "surround"):
setattr(receiver, f"trim_{_trim}_range", NumericRange(-10.0, 10.0, 0.1))
receiver.zone_b_power_on = False
receiver.zone_b_volume = -40.0
receiver.zone_b_mute_enabled = False
receiver.zone_b_source = None
receiver.zone_b_available_sources = []
receiver.zone_b_audio_input = "aux"
receiver.zone_b_streaming_source = "DLNA"
mock_create_receiver.return_value = receiver
return receiver
@pytest.fixture
tests/components/lyngdorf/test_config_flow.py
+8 -2
View File
@@ -12,6 +12,7 @@ from homeassistant.config_entries import SOURCE_SSDP, SOURCE_USER
from homeassistant.const import CONF_HOST
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResultType
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.service_info.ssdp import (
ATTR_UPNP_FRIENDLY_NAME,
ATTR_UPNP_MODEL_NAME,
@@ -26,8 +27,10 @@ pytestmark = pytest.mark.usefixtures("mock_setup_entry")
MOCK_SERIAL = "0050c27c76b2"
@pytest.mark.usefixtures("mock_find_receiver_model", "mock_get_device_serial")
async def test_user_flow(hass: HomeAssistant) -> None:
@pytest.mark.usefixtures("mock_find_receiver_model")
async def test_user_flow(
hass: HomeAssistant, mock_get_device_serial: AsyncMock
) -> None:
"""Test the user configuration flow with serial lookup."""
result = await hass.config_entries.flow.async_init(
DOMAIN,
@@ -48,6 +51,9 @@ async def test_user_flow(hass: HomeAssistant) -> None:
assert config_entry.data[CONF_HOST] == "192.168.1.100"
assert config_entry.data[CONF_SERIAL_NUMBER] == MOCK_SERIAL
assert config_entry.title == "mp-60"
assert mock_get_device_serial.call_args.kwargs[
"session"
] is async_get_clientsession(hass)
@pytest.mark.usefixtures("mock_find_receiver_model")
tests/components/lyngdorf/test_init.py
+12
View File
@@ -10,6 +10,7 @@ from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import CONF_HOST, CONF_MODEL, EVENT_HOMEASSISTANT_STOP
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from tests.common import MockConfigEntry
@@ -69,6 +70,17 @@ async def test_unload_entry(
assert init_integration.state is ConfigEntryState.NOT_LOADED
async def test_uses_the_home_assistant_websession(
hass: HomeAssistant,
init_integration: MockConfigEntry,
mock_create_receiver: MagicMock,
) -> None:
"""Test the receiver is given Home Assistant's shared aiohttp session."""
assert mock_create_receiver.call_args.kwargs["session"] is async_get_clientsession(
hass
)
async def test_unload_releases_receiver_subscriptions(
hass: HomeAssistant,
init_integration: MockConfigEntry,