Address review feedback

- Remove unnecessary int() cast from NumberSelector value
- Remove unreachable invalid source guard in async_select_source
- Use freezer fixture for time-dependent test

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.strict-typing

@@ -337,6 +337,7 @@ homeassistant.components.led_ble.*
 homeassistant.components.lektrico.*
 homeassistant.components.letpot.*
 homeassistant.components.lg_infrared.*
+homeassistant.components.lg_tv_rs232.*
 homeassistant.components.libre_hardware_monitor.*
 homeassistant.components.lidarr.*
 homeassistant.components.liebherr.*

CODEOWNERS

@@ -987,6 +987,8 @@ CLAUDE.md @home-assistant/core
 /tests/components/lg_netcast/ @Drafteed @splinter98
 /homeassistant/components/lg_thinq/ @LG-ThinQ-Integration
 /tests/components/lg_thinq/ @LG-ThinQ-Integration
+/homeassistant/components/lg_tv_rs232/ @balloob
+/tests/components/lg_tv_rs232/ @balloob
 /homeassistant/components/libre_hardware_monitor/ @Sab44
 /tests/components/libre_hardware_monitor/ @Sab44
 /homeassistant/components/lichess/ @aryanhasgithub

homeassistant/components/lg_tv_rs232/__init__.py

@@ -0,0 +1,52 @@
+"""The LG TV RS-232 integration."""
+
+from lg_rs232_tv import LGTV, TVState
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import CONF_DEVICE, Platform
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import ConfigEntryNotReady
+
+from .const import CONF_SET_ID, LOGGER, QUERY_ATTRIBUTES, LGTVRS232ConfigEntry
+
+PLATFORMS = [Platform.MEDIA_PLAYER]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: LGTVRS232ConfigEntry) -> bool:
+    """Set up LG TV RS-232 from a config entry."""
+    port = entry.data[CONF_DEVICE]
+    tv = LGTV(port, set_id=entry.data[CONF_SET_ID])
+
+    try:
+        await tv.connect()
+        await tv.query(QUERY_ATTRIBUTES)
+    except (ConnectionError, OSError, TimeoutError) as err:
+        if tv.connected:
+            await tv.disconnect()
+        raise ConfigEntryNotReady(f"Error connecting to LG TV: {err}") from err
+
+    entry.runtime_data = tv
+
+    @callback
+    def _on_disconnect(state: TVState | None) -> None:
+        # Only reload if the entry is still loaded. During entry removal,
+        # disconnect() fires this callback but the entry is already gone.
+        if state is None and entry.state is ConfigEntryState.LOADED:
+            LOGGER.warning("LG TV disconnected, reloading config entry")
+            hass.config_entries.async_schedule_reload(entry.entry_id)
+
+    entry.async_on_unload(tv.subscribe(_on_disconnect))
+
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: LGTVRS232ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+    if unload_ok:
+        await entry.runtime_data.disconnect()
+
+    return unload_ok

homeassistant/components/lg_tv_rs232/config_flow.py

@@ -0,0 +1,102 @@
+"""Config flow for the LG TV RS-232 integration."""
+
+from typing import Any
+
+from lg_rs232_tv import DEFAULT_SET_ID, LGTV, TVNotRespondingError
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_DEVICE
+from homeassistant.helpers.selector import (
+    NumberSelector,
+    NumberSelectorConfig,
+    NumberSelectorMode,
+    SerialPortSelector,
+)
+
+from .const import CONF_SET_ID, DOMAIN, LOGGER
+
+DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_DEVICE): SerialPortSelector(),
+        vol.Required(CONF_SET_ID, default=DEFAULT_SET_ID): NumberSelector(
+            NumberSelectorConfig(min=1, max=99, mode=NumberSelectorMode.BOX)
+        ),
+    }
+)
+
+# Outcome of _async_attempt_connect that means the serial port works but no LG
+# TV answered it; this routes the user to the troubleshooting step.
+RESULT_NO_TV = "no_tv"
+
+
+async def _async_attempt_connect(port: str, set_id: int) -> str | None:
+    """Attempt to connect to the TV at the given port.
+
+    Returns None on success, otherwise an outcome key: "cannot_connect" when
+    the serial port could not be opened, RESULT_NO_TV when the port works but
+    no LG TV responded to it, or "unknown" for an unexpected error.
+    """
+    tv = LGTV(port, set_id=set_id)
+
+    try:
+        await tv.connect()
+    except TVNotRespondingError:
+        # The port was opened but no LG TV responded to the power query.
+        return RESULT_NO_TV
+    except ValueError, ConnectionError, OSError, TimeoutError:
+        # The serial port itself could not be opened.
+        return "cannot_connect"
+    except Exception:  # noqa: BLE001
+        LOGGER.exception("Unexpected exception")
+        return "unknown"
+    else:
+        await tv.disconnect()
+        return None
+
+
+class LGTVRS232ConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for LG TV RS-232."""
+
+    VERSION = 1
+
+    _user_input: dict[str, Any] | None = None
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            port = user_input[CONF_DEVICE]
+            set_id = user_input[CONF_SET_ID]
+
+            self._async_abort_entries_match({CONF_DEVICE: port, CONF_SET_ID: set_id})
+            error = await _async_attempt_connect(port, set_id)
+            if error is None:
+                return self.async_create_entry(
+                    title="LG TV",
+                    data={CONF_DEVICE: port, CONF_SET_ID: set_id},
+                )
+            if error == RESULT_NO_TV:
+                self._user_input = user_input
+                return await self.async_step_troubleshoot()
+            errors["base"] = error
+
+        return self.async_show_form(
+            step_id="user",
+            data_schema=self.add_suggested_values_to_schema(
+                DATA_SCHEMA, user_input or self._user_input or {}
+            ),
+            errors=errors,
+        )
+
+    async def async_step_troubleshoot(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Guide the user to enable RS-232 control after a failed connection."""
+        if user_input is not None:
+            return await self.async_step_user()
+
+        return self.async_show_form(step_id="troubleshoot")

homeassistant/components/lg_tv_rs232/const.py

@@ -0,0 +1,18 @@
+"""Constants for the LG TV RS-232 integration."""
+
+import logging
+
+from lg_rs232_tv import LGTV
+
+from homeassistant.config_entries import ConfigEntry
+
+LOGGER = logging.getLogger(__package__)
+DOMAIN = "lg_tv_rs232"
+
+CONF_SET_ID = "set_id"
+
+# TVState attributes the integration polls for; the TV is not asked for
+# attributes the media player entity does not use.
+QUERY_ATTRIBUTES = ("power", "input_source", "volume", "volume_mute", "balance")
+
+type LGTVRS232ConfigEntry = ConfigEntry[LGTV]

homeassistant/components/lg_tv_rs232/manifest.json

@@ -0,0 +1,13 @@
+{
+  "domain": "lg_tv_rs232",
+  "name": "LG TV via Serial",
+  "codeowners": ["@balloob"],
+  "config_flow": true,
+  "dependencies": ["usb"],
+  "documentation": "https://www.home-assistant.io/integrations/lg_tv_rs232",
+  "integration_type": "device",
+  "iot_class": "local_polling",
+  "loggers": ["lg_rs232_tv"],
+  "quality_scale": "silver",
+  "requirements": ["lg-rs232-tv==1.2.0"]
+}

homeassistant/components/lg_tv_rs232/media_player.py

@@ -0,0 +1,186 @@
+"""Media player platform for the LG TV RS-232 integration."""
+
+from collections.abc import Callable, Coroutine
+from datetime import timedelta
+from functools import wraps
+from typing import Any
+
+from lg_rs232_tv import MAX_VOLUME, CommandRejected, InputSource, PowerState, TVState
+
+from homeassistant.components.media_player import (
+    MediaPlayerDeviceClass,
+    MediaPlayerEntity,
+    MediaPlayerEntityFeature,
+    MediaPlayerState,
+)
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import DOMAIN, QUERY_ATTRIBUTES, LGTVRS232ConfigEntry
+
+# LG TVs do not push state over RS-232, so the entity is polled.
+SCAN_INTERVAL = timedelta(seconds=5)
+PARALLEL_UPDATES = 1
+
+INPUT_SOURCE_LG_TO_HA: dict[InputSource, str] = {
+    InputSource.DTV_ANTENNA: "dtv_antenna",
+    InputSource.DTV_CABLE: "dtv_cable",
+    InputSource.ANALOG_ANTENNA: "analog_antenna",
+    InputSource.ANALOG_CABLE: "analog_cable",
+    InputSource.AV1: "av1",
+    InputSource.AV2: "av2",
+    InputSource.COMPONENT1: "component1",
+    InputSource.COMPONENT2: "component2",
+    InputSource.COMPONENT3: "component3",
+    InputSource.RGB_PC: "rgb_pc",
+    InputSource.HDMI1: "hdmi1",
+    InputSource.HDMI2: "hdmi2",
+    InputSource.HDMI3: "hdmi3",
+    InputSource.HDMI4: "hdmi4",
+}
+INPUT_SOURCE_HA_TO_LG: dict[str, InputSource] = {
+    value: key for key, value in INPUT_SOURCE_LG_TO_HA.items()
+}
+
+_BASE_SUPPORTED_FEATURES = (
+    MediaPlayerEntityFeature.TURN_ON
+    | MediaPlayerEntityFeature.TURN_OFF
+    | MediaPlayerEntityFeature.SELECT_SOURCE
+)
+
+
+def catch_command_errors[**_P](
+    func: Callable[_P, Coroutine[Any, Any, None]],
+) -> Callable[_P, Coroutine[Any, Any, None]]:
+    """Translate LG library errors raised by an action into HomeAssistantError."""
+
+    @wraps(func)
+    async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
+        try:
+            await func(*args, **kwargs)
+        except CommandRejected as err:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="command_rejected",
+                translation_placeholders={"error": str(err)},
+            ) from err
+        except (ConnectionError, OSError, TimeoutError) as err:
+            raise HomeAssistantError(
+                translation_domain=DOMAIN,
+                translation_key="command_failed",
+                translation_placeholders={"error": str(err)},
+            ) from err
+
+    return wrapper
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    config_entry: LGTVRS232ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up the LG TV RS-232 media player."""
+    async_add_entities([LGTVRS232MediaPlayer(config_entry)])
+
+
+class LGTVRS232MediaPlayer(MediaPlayerEntity):
+    """Representation of an LG TV controlled over RS-232."""
+
+    _attr_device_class = MediaPlayerDeviceClass.TV
+    _attr_has_entity_name = True
+    _attr_name = None
+    _attr_translation_key = "tv"
+    _attr_source_list = sorted(INPUT_SOURCE_LG_TO_HA.values())
+
+    def __init__(self, config_entry: LGTVRS232ConfigEntry) -> None:
+        """Initialize the media player."""
+        self._tv = config_entry.runtime_data
+        self._attr_unique_id = config_entry.entry_id
+        self._attr_device_info = DeviceInfo(
+            identifiers={(DOMAIN, config_entry.entry_id)},
+            manufacturer="LG",
+        )
+        self._async_update_from_state(self._tv.state)
+
+    async def async_added_to_hass(self) -> None:
+        """Subscribe to TV state updates."""
+        self.async_on_remove(self._tv.subscribe(self._async_on_state_update))
+
+    async def async_update(self) -> None:
+        """Poll the TV for its current state."""
+        await self._tv.query(QUERY_ATTRIBUTES)
+
+    @callback
+    def _async_on_state_update(self, state: TVState | None) -> None:
+        """Handle a state update from the TV."""
+        if state is None:
+            self._attr_available = False
+        else:
+            self._attr_available = True
+            self._async_update_from_state(state)
+        self.async_write_ha_state()
+
+    @callback
+    def _async_update_from_state(self, state: TVState) -> None:
+        """Update entity attributes from a TV state snapshot."""
+        if state.power is None:
+            self._attr_state = None
+        else:
+            self._attr_state = (
+                MediaPlayerState.ON
+                if state.power is PowerState.ON
+                else MediaPlayerState.OFF
+            )
+
+        source = state.input_source
+        self._attr_source = INPUT_SOURCE_LG_TO_HA.get(source) if source else None
+
+        # The TV only answers the balance query when its own speaker is the
+        # active audio output. When audio is routed elsewhere (e.g. optical),
+        # the TV's volume does not reflect what the user hears, so neither the
+        # volume controls nor the volume attributes are exposed.
+        features = _BASE_SUPPORTED_FEATURES
+        if state.balance is None:
+            self._attr_volume_level = None
+            self._attr_is_volume_muted = None
+        else:
+            features |= (
+                MediaPlayerEntityFeature.VOLUME_SET
+                | MediaPlayerEntityFeature.VOLUME_STEP
+                | MediaPlayerEntityFeature.VOLUME_MUTE
+            )
+            self._attr_volume_level = (
+                None if state.volume is None else state.volume / MAX_VOLUME
+            )
+            self._attr_is_volume_muted = state.volume_mute
+        self._attr_supported_features = features
+
+    @catch_command_errors
+    async def async_turn_on(self) -> None:
+        """Turn the TV on."""
+        await self._tv.power_on()
+
+    @catch_command_errors
+    async def async_turn_off(self) -> None:
+        """Turn the TV off."""
+        await self._tv.power_off()
+
+    @catch_command_errors
+    async def async_set_volume_level(self, volume: float) -> None:
+        """Set volume level, range 0..1."""
+        await self._tv.set_volume(round(volume * MAX_VOLUME))
+
+    @catch_command_errors
+    async def async_mute_volume(self, mute: bool) -> None:
+        """Mute or unmute the TV."""
+        if mute:
+            await self._tv.mute_on()
+        else:
+            await self._tv.mute_off()
+
+    @catch_command_errors
+    async def async_select_source(self, source: str) -> None:
+        """Select an input source."""
+        await self._tv.select_input_source(INPUT_SOURCE_HA_TO_LG[source])

homeassistant/components/lg_tv_rs232/quality_scale.yaml

@@ -0,0 +1,84 @@
+rules:
+  # Bronze
+  action-setup:
+    status: exempt
+    comment: "The integration does not register custom actions."
+  appropriate-polling: done
+  brands: done
+  common-modules: done
+  config-flow-test-coverage: done
+  config-flow: done
+  dependency-transparency: done
+  docs-actions:
+    status: exempt
+    comment: "The integration does not register custom actions."
+  docs-high-level-description: done
+  docs-installation-instructions: done
+  docs-removal-instructions: done
+  entity-event-setup: done
+  entity-unique-id: done
+  has-entity-name: done
+  runtime-data: done
+  test-before-configure: done
+  test-before-setup: done
+  unique-config-entry: done
+
+  # Silver
+  action-exceptions: done
+  config-entry-unloading: done
+  docs-configuration-parameters:
+    status: exempt
+    comment: "The integration has no options to configure."
+  docs-installation-parameters: done
+  entity-unavailable: done
+  integration-owner: done
+  log-when-unavailable: done
+  parallel-updates: done
+  reauthentication-flow:
+    status: exempt
+    comment: "The integration does not require authentication."
+  test-coverage: done
+
+  # Gold
+  devices: done
+  diagnostics: todo
+  discovery-update-info:
+    status: exempt
+    comment: "Serial devices are configured manually; there is no discovery."
+  discovery:
+    status: exempt
+    comment: "RS-232 serial connections cannot be discovered."
+  docs-data-update: done
+  docs-examples: done
+  docs-known-limitations: done
+  docs-supported-devices: done
+  docs-supported-functions: done
+  docs-troubleshooting: done
+  docs-use-cases: done
+  dynamic-devices:
+    status: exempt
+    comment: "The integration does not create dynamic devices."
+  entity-category: done
+  entity-device-class: done
+  entity-disabled-by-default:
+    status: exempt
+    comment: "The integration only provides a single primary entity."
+  entity-translations: done
+  exception-translations: done
+  icon-translations:
+    status: exempt
+    comment: "The media player entity uses its device class for its icon."
+  reconfiguration-flow: todo
+  repair-issues:
+    status: exempt
+    comment: "The integration has no user-actionable issues to repair."
+  stale-devices:
+    status: exempt
+    comment: "The integration does not create devices that can become stale."
+
+  # Platinum
+  async-dependency: done
+  inject-websession:
+    status: exempt
+    comment: "The integration does not make HTTP requests."
+  strict-typing: done

homeassistant/components/lg_tv_rs232/strings.json

@@ -0,0 +1,61 @@
+{
+  "config": {
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "step": {
+      "troubleshoot": {
+        "description": "Home Assistant could not communicate with the LG TV over the serial port.\n\nThe most common cause is that **RS-232C Control** is not enabled on the TV. On most LG models this setting is in a hidden service menu (often called **InStart**); the exact location varies by model, so check your TV's documentation.\n\nAlso make sure that:\n- The TV is powered on.\n- The serial cable is a null-modem (cross-over) cable and is fully seated. LG's RS-232 jack is recessed, so push the plug in until it clicks.\n- The correct serial port was selected.\n\nSelect **Submit** to try again.",
+        "title": "Connection failed"
+      },
+      "user": {
+        "data": {
+          "device": "[%key:common::config_flow::data::port%]",
+          "set_id": "Set ID"
+        },
+        "data_description": {
+          "device": "Serial port path to connect to. The TV must be powered on for the initial connection.",
+          "set_id": "The set ID configured on the TV. Leave this at 1 unless you have multiple TVs daisy-chained on the same RS-232 bus."
+        }
+      }
+    }
+  },
+  "entity": {
+    "media_player": {
+      "tv": {
+        "state_attributes": {
+          "source": {
+            "state": {
+              "analog_antenna": "Analog (antenna)",
+              "analog_cable": "Analog (cable)",
+              "av1": "AV 1",
+              "av2": "AV 2",
+              "component1": "Component 1",
+              "component2": "Component 2",
+              "component3": "Component 3",
+              "dtv_antenna": "Digital TV (antenna)",
+              "dtv_cable": "Digital TV (cable)",
+              "hdmi1": "HDMI 1",
+              "hdmi2": "HDMI 2",
+              "hdmi3": "HDMI 3",
+              "hdmi4": "HDMI 4",
+              "rgb_pc": "RGB PC"
+            }
+          }
+        }
+      }
+    }
+  },
+  "exceptions": {
+    "command_failed": {
+      "message": "Failed to send the command to the TV: {error}"
+    },
+    "command_rejected": {
+      "message": "The TV rejected the command: {error}"
+    }
+  }
+}

homeassistant/generated/config_flows.py

@@ -404,6 +404,7 @@ FLOWS = {
         "lg_netcast",
         "lg_soundbar",
         "lg_thinq",
+        "lg_tv_rs232",
         "libre_hardware_monitor",
         "lichess",
         "lidarr",

homeassistant/generated/integrations.json

@@ -3767,6 +3767,12 @@
         }
       }
     },
+    "lg_tv_rs232": {
+      "name": "LG TV via Serial",
+      "integration_type": "device",
+      "config_flow": true,
+      "iot_class": "local_polling"
+    },
     "libre_hardware_monitor": {
       "name": "Libre Hardware Monitor",
       "integration_type": "device",

mypy.ini

@@ -3127,6 +3127,16 @@ disallow_untyped_defs = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.lg_tv_rs232.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.libre_hardware_monitor.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true

requirements_all.txt

@@ -1455,6 +1455,9 @@ lektricowifi==0.1
 # homeassistant.components.letpot
 letpot==0.7.0
 
+# homeassistant.components.lg_tv_rs232
+lg-rs232-tv==1.2.0
+
 # homeassistant.components.foscam
 libpyfoscamcgi==0.0.9
 

requirements_test_all.txt

@@ -1295,6 +1295,9 @@ lektricowifi==0.1
 # homeassistant.components.letpot
 letpot==0.7.0
 
+# homeassistant.components.lg_tv_rs232
+lg-rs232-tv==1.2.0
+
 # homeassistant.components.foscam
 libpyfoscamcgi==0.0.9
 

tests/components/lg_tv_rs232/__init__.py

@@ -0,0 +1,4 @@
+"""Tests for the LG TV RS-232 integration."""
+
+MOCK_DEVICE = "/dev/ttyUSB0"
+MOCK_SET_ID = 1

tests/components/lg_tv_rs232/conftest.py

@@ -0,0 +1,107 @@
+"""Test fixtures for the LG TV RS-232 integration."""
+
+from unittest.mock import AsyncMock, patch
+
+from lg_rs232_tv import LGTV, InputSource, PowerState, TVState
+import pytest
+
+from homeassistant.components.lg_tv_rs232.const import CONF_SET_ID, DOMAIN
+from homeassistant.const import CONF_DEVICE
+from homeassistant.core import HomeAssistant
+
+from . import MOCK_DEVICE, MOCK_SET_ID
+
+from tests.common import MockConfigEntry
+
+
+class MockLGTV(LGTV):
+    """LG TV test double built on the real controller object."""
+
+    def __init__(self, state: TVState) -> None:
+        """Initialize the mock TV."""
+        super().__init__(MOCK_DEVICE, set_id=MOCK_SET_ID)
+        self._connected = True
+        self._state = state
+        self.connect = AsyncMock(side_effect=self._mock_connect)
+        self.query = AsyncMock()
+        self.disconnect = AsyncMock(side_effect=self._mock_disconnect)
+        self.power_on = AsyncMock()
+        self.power_off = AsyncMock()
+        self.set_volume = AsyncMock()
+        self.mute_on = AsyncMock()
+        self.mute_off = AsyncMock()
+        self.select_input_source = AsyncMock()
+
+    def mock_state(self, state: TVState | None) -> None:
+        """Push a state update through the TV."""
+        self._connected = state is not None
+        if state is not None:
+            self._state = state
+        self._notify_subscribers()
+
+    async def _mock_connect(self) -> None:
+        """Pretend to open the serial connection."""
+        self._connected = True
+
+    async def _mock_disconnect(self) -> None:
+        """Pretend to close the serial connection."""
+        self._connected = False
+        self._notify_subscribers()
+
+
+def _default_state() -> TVState:
+    """Return a TVState with typical defaults."""
+    return TVState(
+        power=PowerState.ON,
+        input_source=InputSource.HDMI1,
+        volume=20,
+        volume_mute=False,
+        balance=50,
+    )
+
+
+@pytest.fixture
+def initial_tv_state() -> TVState:
+    """Return the initial TV state for a test."""
+    return _default_state()
+
+
+@pytest.fixture
+def mock_lgtv(initial_tv_state: TVState) -> MockLGTV:
+    """Create a mock LGTV controller."""
+    return MockLGTV(initial_tv_state)
+
+
+@pytest.fixture
+def mock_config_entry(hass: HomeAssistant) -> MockConfigEntry:
+    """Create a mock config entry."""
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID},
+        title="LG TV",
+        entry_id="01KPBBPM6WCQ8148EFR0TCG1WW",
+    )
+    entry.add_to_hass(hass)
+    return entry
+
+
+@pytest.fixture(autouse=True)
+async def mock_usb_component(hass: HomeAssistant) -> None:
+    """Mock the USB component to prevent setup failures."""
+    hass.config.components.add("usb")
+
+
+@pytest.fixture
+async def init_components(
+    hass: HomeAssistant,
+    mock_usb_component: None,
+    mock_lgtv: MockLGTV,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Initialize the LG TV RS-232 component."""
+    with patch(
+        "homeassistant.components.lg_tv_rs232.LGTV",
+        return_value=mock_lgtv,
+    ):
+        await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()

tests/components/lg_tv_rs232/snapshots/test_media_player.ambr

@@ -0,0 +1,89 @@
+# serializer version: 1
+# name: test_entities_created[media_player.lg_tv-entry]
+  EntityRegistryEntrySnapshot({
+    'aliases': list([
+      None,
+    ]),
+    'area_id': None,
+    'capabilities': dict({
+      'source_list': list([
+        'analog_antenna',
+        'analog_cable',
+        'av1',
+        'av2',
+        'component1',
+        'component2',
+        'component3',
+        'dtv_antenna',
+        'dtv_cable',
+        'hdmi1',
+        'hdmi2',
+        'hdmi3',
+        'hdmi4',
+        'rgb_pc',
+      ]),
+    }),
+    'config_entry_id': <ANY>,
+    'config_subentry_id': <ANY>,
+    'device_class': None,
+    'device_id': <ANY>,
+    'disabled_by': None,
+    'domain': 'media_player',
+    'entity_category': None,
+    'entity_id': 'media_player.lg_tv',
+    'has_entity_name': True,
+    'hidden_by': None,
+    'icon': None,
+    'id': <ANY>,
+    'labels': set({
+    }),
+    'name': None,
+    'object_id_base': None,
+    'options': dict({
+    }),
+    'original_device_class': <MediaPlayerDeviceClass.TV: 'tv'>,
+    'original_icon': None,
+    'original_name': None,
+    'platform': 'lg_tv_rs232',
+    'previous_unique_id': None,
+    'suggested_object_id': None,
+    'supported_features': <MediaPlayerEntityFeature: 3468>,
+    'translation_key': 'tv',
+    'unique_id': '01KPBBPM6WCQ8148EFR0TCG1WW',
+    'unit_of_measurement': None,
+  })
+# ---
+# name: test_entities_created[media_player.lg_tv-state]
+  StateSnapshot({
+    'attributes': ReadOnlyDict({
+      'device_class': 'tv',
+      'friendly_name': 'LG TV',
+      'is_volume_muted': False,
+      'source': 'hdmi1',
+      'source_list': list([
+        'analog_antenna',
+        'analog_cable',
+        'av1',
+        'av2',
+        'component1',
+        'component2',
+        'component3',
+        'dtv_antenna',
+        'dtv_cable',
+        'hdmi1',
+        'hdmi2',
+        'hdmi3',
+        'hdmi4',
+        'rgb_pc',
+      ]),
+      'supported_features': <MediaPlayerEntityFeature: 3468>,
+      'volume_level': 0.2,
+    }),
+    'context': <ANY>,
+    'entity_id': 'media_player.lg_tv',
+    'last_changed': <ANY>,
+    'last_reported': <ANY>,
+    'last_updated': <ANY>,
+    'state': 'on',
+  })
+# ---

tests/components/lg_tv_rs232/test_config_flow.py

@@ -0,0 +1,166 @@
+"""Tests for the LG TV RS-232 config flow."""
+
+from collections.abc import Generator
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from lg_rs232_tv import TVNotRespondingError
+import pytest
+
+from homeassistant.components.lg_tv_rs232.const import CONF_SET_ID, DOMAIN
+from homeassistant.config_entries import SOURCE_USER
+from homeassistant.const import CONF_DEVICE
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResultType
+
+from . import MOCK_DEVICE, MOCK_SET_ID
+
+from tests.common import MockConfigEntry
+
+
+@pytest.fixture
+def mock_async_setup_entry(mock_lgtv: MagicMock) -> Generator[AsyncMock]:
+    """Prevent config-entry creation tests from setting up the integration."""
+
+    with patch(
+        "homeassistant.components.lg_tv_rs232.async_setup_entry",
+        return_value=True,
+    ) as mock_setup_entry:
+        yield mock_setup_entry
+
+
+async def test_user_form_creates_entry(
+    hass: HomeAssistant,
+    mock_lgtv: MagicMock,
+    mock_async_setup_entry: AsyncMock,
+) -> None:
+    """Test successful config flow creates an entry."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    with patch(
+        "homeassistant.components.lg_tv_rs232.config_flow.LGTV",
+        return_value=mock_lgtv,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID},
+        )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert result["title"] == "LG TV"
+    assert result["data"] == {CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID}
+    mock_async_setup_entry.assert_awaited_once()
+    mock_lgtv.connect.assert_awaited_once()
+    mock_lgtv.disconnect.assert_awaited_once()
+
+
+async def test_user_form_no_tv_shows_troubleshooting(
+    hass: HomeAssistant,
+    mock_lgtv: MagicMock,
+    mock_async_setup_entry: AsyncMock,
+) -> None:
+    """Test a working port with no LG TV routes to the troubleshooting step."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+    mock_lgtv.connect.side_effect = TVNotRespondingError("No response from LG TV")
+
+    with patch(
+        "homeassistant.components.lg_tv_rs232.config_flow.LGTV",
+        return_value=mock_lgtv,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID},
+        )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "troubleshoot"
+
+    # Continuing from troubleshooting returns to the user step to retry.
+    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+
+    mock_lgtv.connect.side_effect = None
+
+    with patch(
+        "homeassistant.components.lg_tv_rs232.config_flow.LGTV",
+        return_value=mock_lgtv,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID},
+        )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+
+@pytest.mark.parametrize(
+    ("exception", "error"),
+    [
+        (ValueError("Invalid port"), "cannot_connect"),
+        (OSError("No such device"), "cannot_connect"),
+        (ConnectionRefusedError("Connection refused"), "cannot_connect"),
+        (RuntimeError("boom"), "unknown"),
+    ],
+)
+async def test_user_form_bad_port_shows_error(
+    hass: HomeAssistant,
+    exception: Exception,
+    error: str,
+    mock_lgtv: MagicMock,
+    mock_async_setup_entry: AsyncMock,
+) -> None:
+    """Test a bad serial port keeps the user on the form with an error."""
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+    mock_lgtv.connect.side_effect = exception
+
+    with patch(
+        "homeassistant.components.lg_tv_rs232.config_flow.LGTV",
+        return_value=mock_lgtv,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID},
+        )
+
+    assert result["type"] is FlowResultType.FORM
+    assert result["step_id"] == "user"
+    assert result["errors"] == {"base": error}
+
+    mock_lgtv.connect.side_effect = None
+
+    with patch(
+        "homeassistant.components.lg_tv_rs232.config_flow.LGTV",
+        return_value=mock_lgtv,
+    ):
+        result = await hass.config_entries.flow.async_configure(
+            result["flow_id"],
+            {CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID},
+        )
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+
+
+async def test_user_duplicate_aborts(hass: HomeAssistant) -> None:
+    """Test we abort if the same port and set ID are already configured."""
+    entry = MockConfigEntry(
+        domain=DOMAIN, data={CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID}
+    )
+    entry.add_to_hass(hass)
+
+    result = await hass.config_entries.flow.async_init(
+        DOMAIN, context={"source": SOURCE_USER}
+    )
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {CONF_DEVICE: MOCK_DEVICE, CONF_SET_ID: MOCK_SET_ID},
+    )
+
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "already_configured"

tests/components/lg_tv_rs232/test_init.py

@@ -0,0 +1,65 @@
+"""Tests for the LG TV RS-232 integration init."""
+
+from unittest.mock import AsyncMock, patch
+
+from homeassistant.config_entries import ConfigEntryState
+from homeassistant.core import HomeAssistant
+
+from .conftest import MockLGTV
+
+from tests.common import MockConfigEntry
+
+
+async def test_setup_connection_error(
+    hass: HomeAssistant,
+    mock_lgtv: MockLGTV,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test a connection failure results in a retry."""
+    mock_lgtv.connect.side_effect = TimeoutError
+
+    with patch(
+        "homeassistant.components.lg_tv_rs232.LGTV",
+        return_value=mock_lgtv,
+    ):
+        await hass.config_entries.async_setup(mock_config_entry.entry_id)
+        await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
+
+
+async def test_unload_entry(
+    hass: HomeAssistant,
+    mock_lgtv: MockLGTV,
+    init_components: None,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test unloading a config entry disconnects from the TV."""
+    assert mock_config_entry.state is ConfigEntryState.LOADED
+
+    await hass.config_entries.async_unload(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
+    mock_lgtv.disconnect.assert_awaited_once()
+
+
+async def test_remove_entry_while_loaded(
+    hass: HomeAssistant,
+    mock_lgtv: MockLGTV,
+    init_components: AsyncMock,
+    mock_config_entry: MockConfigEntry,
+) -> None:
+    """Test removing a config entry does not schedule a reload.
+
+    When removing a loaded entry, disconnect() fires the subscriber callback
+    with state=None. The callback must not schedule a reload because the entry
+    is already being removed (state is no longer LOADED).
+    """
+    assert mock_config_entry.state is ConfigEntryState.LOADED
+
+    await hass.config_entries.async_remove(mock_config_entry.entry_id)
+    await hass.async_block_till_done()
+
+    assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
+    mock_lgtv.disconnect.assert_awaited_once()

tests/components/lg_tv_rs232/test_media_player.py

@@ -0,0 +1,253 @@
+"""Tests for the LG TV RS-232 media player platform."""
+
+from pathlib import Path
+from unittest.mock import call
+
+from freezegun.api import FrozenDateTimeFactory
+from lg_rs232_tv import CommandRejected, InputSource, PowerState, TVState
+import pytest
+from syrupy.assertion import SnapshotAssertion
+
+from homeassistant.components.lg_tv_rs232.media_player import (
+    INPUT_SOURCE_LG_TO_HA,
+    SCAN_INTERVAL,
+)
+from homeassistant.components.media_player import (
+    ATTR_INPUT_SOURCE,
+    ATTR_INPUT_SOURCE_LIST,
+    ATTR_MEDIA_VOLUME_LEVEL,
+    ATTR_MEDIA_VOLUME_MUTED,
+    DOMAIN as MP_DOMAIN,
+    SERVICE_SELECT_SOURCE,
+    MediaPlayerEntityFeature,
+)
+from homeassistant.const import (
+    ATTR_ENTITY_ID,
+    ATTR_SUPPORTED_FEATURES,
+    SERVICE_TURN_OFF,
+    SERVICE_TURN_ON,
+    SERVICE_VOLUME_DOWN,
+    SERVICE_VOLUME_MUTE,
+    SERVICE_VOLUME_SET,
+    SERVICE_VOLUME_UP,
+    STATE_OFF,
+    STATE_ON,
+    STATE_UNAVAILABLE,
+    STATE_UNKNOWN,
+)
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import entity_registry as er
+from homeassistant.util.json import load_json
+
+from .conftest import MockLGTV, _default_state
+
+from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
+
+ENTITY_ID = "media_player.lg_tv"
+
+
+@pytest.fixture(autouse=True)
+async def auto_init_components(init_components: None) -> None:
+    """Set up the component."""
+
+
+async def test_entities_created(
+    hass: HomeAssistant,
+    mock_lgtv: MockLGTV,
+    snapshot: SnapshotAssertion,
+    mock_config_entry: MockConfigEntry,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Test the media player entity is created through config entry setup."""
+    await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
+    mock_lgtv.query.assert_awaited()
+
+
+async def test_polling_updates_state(
+    hass: HomeAssistant, mock_lgtv: MockLGTV, freezer: FrozenDateTimeFactory
+) -> None:
+    """Test the entity polls the TV on the scan interval and reflects changes."""
+    assert hass.states.get(ENTITY_ID).state == STATE_ON
+
+    off_state = _default_state()
+    off_state.power = PowerState.OFF
+    mock_lgtv.query.side_effect = lambda _attrs: mock_lgtv.mock_state(off_state)
+
+    freezer.tick(SCAN_INTERVAL)
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    mock_lgtv.query.assert_awaited()
+    assert hass.states.get(ENTITY_ID).state == STATE_OFF
+
+
+async def test_state_updates(hass: HomeAssistant, mock_lgtv: MockLGTV) -> None:
+    """Test the entity updates from TV pushes and disconnects."""
+    assert hass.states.get(ENTITY_ID).state == STATE_ON
+
+    state = _default_state()
+    state.power = PowerState.OFF
+    mock_lgtv.mock_state(state)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(ENTITY_ID).state == STATE_OFF
+
+    mock_lgtv.mock_state(None)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(ENTITY_ID).state == STATE_UNAVAILABLE
+
+
+async def test_state_unknown_when_not_queried(
+    hass: HomeAssistant, mock_lgtv: MockLGTV
+) -> None:
+    """Test attributes are cleared when the TV reports no state."""
+    mock_lgtv.mock_state(TVState())
+    await hass.async_block_till_done()
+
+    state = hass.states.get(ENTITY_ID)
+    assert state.state == STATE_UNKNOWN
+    assert state.attributes.get(ATTR_MEDIA_VOLUME_LEVEL) is None
+
+
+async def test_power_controls(hass: HomeAssistant, mock_lgtv: MockLGTV) -> None:
+    """Test power services call the right methods."""
+    await hass.services.async_call(
+        MP_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
+    )
+    mock_lgtv.power_on.assert_awaited_once()
+
+    await hass.services.async_call(
+        MP_DOMAIN, SERVICE_TURN_OFF, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
+    )
+    mock_lgtv.power_off.assert_awaited_once()
+
+
+async def test_volume_controls(hass: HomeAssistant, mock_lgtv: MockLGTV) -> None:
+    """Test volume state and controls."""
+    assert hass.states.get(ENTITY_ID).attributes[ATTR_MEDIA_VOLUME_LEVEL] == 0.2
+
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_SET,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_LEVEL: 0.5},
+        blocking=True,
+    )
+    assert mock_lgtv.set_volume.await_args == call(50)
+
+    # Volume up/down use the media player base class default step.
+    await hass.services.async_call(
+        MP_DOMAIN, SERVICE_VOLUME_UP, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
+    )
+    assert mock_lgtv.set_volume.await_args == call(30)
+
+    await hass.services.async_call(
+        MP_DOMAIN, SERVICE_VOLUME_DOWN, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
+    )
+    assert mock_lgtv.set_volume.await_args == call(10)
+
+
+async def test_mute_controls(hass: HomeAssistant, mock_lgtv: MockLGTV) -> None:
+    """Test mute state and controls."""
+    assert hass.states.get(ENTITY_ID).attributes[ATTR_MEDIA_VOLUME_MUTED] is False
+
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_MUTE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: True},
+        blocking=True,
+    )
+    mock_lgtv.mute_on.assert_awaited_once()
+
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_VOLUME_MUTE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_MEDIA_VOLUME_MUTED: False},
+        blocking=True,
+    )
+    mock_lgtv.mute_off.assert_awaited_once()
+
+
+async def test_volume_features_depend_on_speaker_output(
+    hass: HomeAssistant, mock_lgtv: MockLGTV
+) -> None:
+    """Test volume features are dropped when the TV speaker is not the output.
+
+    The TV only answers the balance query when its own speaker is the active
+    audio output, so balance availability gates the volume features.
+    """
+    features = hass.states.get(ENTITY_ID).attributes[ATTR_SUPPORTED_FEATURES]
+    assert features & MediaPlayerEntityFeature.VOLUME_SET
+    assert features & MediaPlayerEntityFeature.VOLUME_STEP
+    assert features & MediaPlayerEntityFeature.VOLUME_MUTE
+
+    state = _default_state()
+    state.balance = None
+    mock_lgtv.mock_state(state)
+    await hass.async_block_till_done()
+
+    entity_state = hass.states.get(ENTITY_ID)
+    features = entity_state.attributes[ATTR_SUPPORTED_FEATURES]
+    assert not features & MediaPlayerEntityFeature.VOLUME_SET
+    assert not features & MediaPlayerEntityFeature.VOLUME_STEP
+    assert not features & MediaPlayerEntityFeature.VOLUME_MUTE
+    assert features & MediaPlayerEntityFeature.SELECT_SOURCE
+
+    # The volume attributes are dropped along with the controls.
+    assert ATTR_MEDIA_VOLUME_LEVEL not in entity_state.attributes
+    assert ATTR_MEDIA_VOLUME_MUTED not in entity_state.attributes
+
+
+async def test_source_state_and_controls(
+    hass: HomeAssistant, mock_lgtv: MockLGTV
+) -> None:
+    """Test source state and selection."""
+    entity_state = hass.states.get(ENTITY_ID)
+    assert entity_state.attributes[ATTR_INPUT_SOURCE] == "hdmi1"
+
+    source_list = entity_state.attributes[ATTR_INPUT_SOURCE_LIST]
+    assert "hdmi1" in source_list
+    assert "av1" in source_list
+    assert source_list == sorted(source_list)
+
+    state = _default_state()
+    state.input_source = InputSource.HDMI2
+    mock_lgtv.mock_state(state)
+    await hass.async_block_till_done()
+
+    assert hass.states.get(ENTITY_ID).attributes[ATTR_INPUT_SOURCE] == "hdmi2"
+
+    await hass.services.async_call(
+        MP_DOMAIN,
+        SERVICE_SELECT_SOURCE,
+        {ATTR_ENTITY_ID: ENTITY_ID, ATTR_INPUT_SOURCE: "hdmi3"},
+        blocking=True,
+    )
+    assert mock_lgtv.select_input_source.await_args == call(InputSource.HDMI3)
+
+
+@pytest.mark.parametrize(
+    "exception",
+    [CommandRejected("a", "02"), ConnectionError("connection lost"), TimeoutError],
+)
+async def test_command_error_raises(
+    hass: HomeAssistant, mock_lgtv: MockLGTV, exception: Exception
+) -> None:
+    """Test library errors raised during an action surface as HomeAssistantError."""
+    mock_lgtv.power_on.side_effect = exception
+
+    with pytest.raises(HomeAssistantError):
+        await hass.services.async_call(
+            MP_DOMAIN, SERVICE_TURN_ON, {ATTR_ENTITY_ID: ENTITY_ID}, blocking=True
+        )
+
+
+def test_input_source_translation_keys_cover_all_enum_members() -> None:
+    """Test all input sources have a declared translation key."""
+    assert set(INPUT_SOURCE_LG_TO_HA) == set(InputSource)
+
+    strings = load_json(Path("homeassistant/components/lg_tv_rs232/strings.json"))
+    assert set(INPUT_SOURCE_LG_TO_HA.values()) == set(
+        strings["entity"]["media_player"]["tv"]["state_attributes"]["source"]["state"]
+    )