diff --git a/homeassistant/components/webostv/media_player.py b/homeassistant/components/webostv/media_player.py index 339124142b1..1d7c92741a8 100644 --- a/homeassistant/components/webostv/media_player.py +++ b/homeassistant/components/webostv/media_player.py @@ -1,14 +1,18 @@ """Support for interface with an LG webOS Smart TV.""" from __future__ import annotations +import asyncio from collections.abc import Awaitable, Callable, Coroutine from contextlib import suppress from datetime import timedelta from functools import wraps +from http import HTTPStatus import logging +from ssl import SSLContext from typing import Any, TypeVar, cast from aiowebostv import WebOsClient, WebOsTvPairError +import async_timeout from typing_extensions import Concatenate, ParamSpec from homeassistant import util @@ -28,6 +32,7 @@ from homeassistant.const import ( ) from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.dispatcher import async_dispatcher_connect from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback @@ -466,3 +471,25 @@ class LgWebOSMediaPlayerEntity(RestoreEntity, MediaPlayerEntity): async def async_command(self, command: str, **kwargs: Any) -> None: """Send a command.""" await self._client.request(command, payload=kwargs.get(ATTR_PAYLOAD)) + + async def _async_fetch_image(self, url: str) -> tuple[bytes | None, str | None]: + """Retrieve an image. + + webOS uses self-signed certificates, thus we need to use an empty + SSLContext to bypass validation errors if url starts with https. + """ + content = None + ssl_context = None + if url.startswith("https"): + ssl_context = SSLContext() + + websession = async_get_clientsession(self.hass) + with suppress(asyncio.TimeoutError), async_timeout.timeout(10): + response = await websession.get(url, ssl=ssl_context) + if response.status == HTTPStatus.OK: + content = await response.read() + + if content is None: + _LOGGER.warning("Error retrieving proxied image from %s", url) + + return content, None diff --git a/tests/components/webostv/test_media_player.py b/tests/components/webostv/test_media_player.py index 70549f5d4e8..e4e2e2ba45f 100644 --- a/tests/components/webostv/test_media_player.py +++ b/tests/components/webostv/test_media_player.py @@ -1,6 +1,7 @@ """The tests for the LG webOS media player platform.""" import asyncio from datetime import timedelta +from http import HTTPStatus from unittest.mock import Mock import pytest @@ -697,3 +698,68 @@ async def test_supported_features_ignore_cache(hass, client): attrs = hass.states.get(ENTITY_ID).attributes assert attrs[ATTR_SUPPORTED_FEATURES] == supported + + +async def test_get_image_http( + hass, client, hass_client_no_auth, aioclient_mock, monkeypatch +): + """Test get image via http.""" + url = "http://something/valid_icon" + monkeypatch.setitem(client.apps[LIVE_TV_APP_ID], "icon", url) + await setup_webostv(hass) + await client.mock_state_update() + + attrs = hass.states.get(ENTITY_ID).attributes + assert "entity_picture_local" not in attrs + + aioclient_mock.get(url, text="image") + client = await hass_client_no_auth() + + resp = await client.get(attrs["entity_picture"]) + content = await resp.read() + + assert content == b"image" + + +async def test_get_image_http_error( + hass, client, hass_client_no_auth, aioclient_mock, caplog, monkeypatch +): + """Test get image via http error.""" + url = "http://something/icon_error" + monkeypatch.setitem(client.apps[LIVE_TV_APP_ID], "icon", url) + await setup_webostv(hass) + await client.mock_state_update() + + attrs = hass.states.get(ENTITY_ID).attributes + assert "entity_picture_local" not in attrs + + aioclient_mock.get(url, exc=asyncio.TimeoutError()) + client = await hass_client_no_auth() + + resp = await client.get(attrs["entity_picture"]) + content = await resp.read() + + assert resp.status == HTTPStatus.INTERNAL_SERVER_ERROR + assert f"Error retrieving proxied image from {url}" in caplog.text + assert content == b"" + + +async def test_get_image_https( + hass, client, hass_client_no_auth, aioclient_mock, monkeypatch +): + """Test get image via http.""" + url = "https://something/valid_icon_https" + monkeypatch.setitem(client.apps[LIVE_TV_APP_ID], "icon", url) + await setup_webostv(hass) + await client.mock_state_update() + + attrs = hass.states.get(ENTITY_ID).attributes + assert "entity_picture_local" not in attrs + + aioclient_mock.get(url, text="https_image") + client = await hass_client_no_auth() + + resp = await client.get(attrs["entity_picture"]) + content = await resp.read() + + assert content == b"https_image"