mirror of
https://github.com/home-assistant/core.git
synced 2025-07-21 20:27:08 +00:00
Add optional base64 decoding of mqtt camera image (#71223)
Add unittest for b64 decoding of camera, fix linting
This commit is contained in:
parent
1a45e54324
commit
0fcdca2d71
@ -1,6 +1,7 @@
|
||||
"""Camera that loads a picture from an MQTT topic."""
|
||||
from __future__ import annotations
|
||||
|
||||
from base64 import b64decode
|
||||
import functools
|
||||
|
||||
import voluptuous as vol
|
||||
@ -16,7 +17,7 @@ from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
|
||||
from . import subscription
|
||||
from .. import mqtt
|
||||
from .const import CONF_QOS, CONF_TOPIC
|
||||
from .const import CONF_ENCODING, CONF_QOS, CONF_TOPIC
|
||||
from .debug_info import log_messages
|
||||
from .mixins import (
|
||||
MQTT_ENTITY_COMMON_SCHEMA,
|
||||
@ -102,7 +103,10 @@ class MqttCamera(MqttEntity, Camera):
|
||||
@log_messages(self.hass, self.entity_id)
|
||||
def message_received(msg):
|
||||
"""Handle new MQTT messages."""
|
||||
self._last_image = msg.payload
|
||||
if self._config[CONF_ENCODING] == "b64":
|
||||
self._last_image = b64decode(msg.payload)
|
||||
else:
|
||||
self._last_image = msg.payload
|
||||
|
||||
self._sub_state = subscription.async_prepare_subscribe_topics(
|
||||
self.hass,
|
||||
|
@ -1,4 +1,5 @@
|
||||
"""The tests for mqtt camera component."""
|
||||
from base64 import b64encode
|
||||
from http import HTTPStatus
|
||||
import json
|
||||
from unittest.mock import patch
|
||||
@ -64,6 +65,34 @@ async def test_run_camera_setup(hass, hass_client_no_auth, mqtt_mock):
|
||||
assert body == "beer"
|
||||
|
||||
|
||||
async def test_run_camera_b64_encoded(hass, hass_client_no_auth, mqtt_mock):
|
||||
"""Test that it fetches the given encoded payload."""
|
||||
topic = "test/camera"
|
||||
await async_setup_component(
|
||||
hass,
|
||||
"camera",
|
||||
{
|
||||
"camera": {
|
||||
"platform": "mqtt",
|
||||
"topic": topic,
|
||||
"name": "Test Camera",
|
||||
"encoding": "b64",
|
||||
}
|
||||
},
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
url = hass.states.get("camera.test_camera").attributes["entity_picture"]
|
||||
|
||||
async_fire_mqtt_message(hass, topic, b64encode(b"grass"))
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get(url)
|
||||
assert resp.status == HTTPStatus.OK
|
||||
body = await resp.text()
|
||||
assert body == "grass"
|
||||
|
||||
|
||||
async def test_availability_when_connection_lost(hass, mqtt_mock):
|
||||
"""Test availability after MQTT disconnection."""
|
||||
await help_test_availability_when_connection_lost(
|
||||
|
Loading…
x
Reference in New Issue
Block a user