Add mqtt encoding support for publishing (#62739)

* encoding support for mqtt publishing - todo tests

* signature allows None values for qos and retain

* common test for mqtt publishing encoding

* better test with command templates

* more tests

* fix tests alarm control panel+tests light basic

* tests light json and template

* add tests vacuum and fix tests light_template

tests/components/mqtt/test_common.py

@@ -8,7 +8,7 @@ from homeassistant.components import mqtt
 from homeassistant.components.mqtt import debug_info
 from homeassistant.components.mqtt.const import MQTT_DISCONNECTED
 from homeassistant.components.mqtt.mixins import MQTT_ATTRIBUTES_BLOCKED
-from homeassistant.const import ATTR_ASSUMED_STATE, STATE_UNAVAILABLE
+from homeassistant.const import ATTR_ASSUMED_STATE, ATTR_ENTITY_ID, STATE_UNAVAILABLE
 from homeassistant.helpers import device_registry as dr, entity_registry as er
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.setup import async_setup_component
@@ -1266,3 +1266,120 @@ async def help_test_entity_category(hass, mqtt_mock, domain, config):
     async_fire_mqtt_message(hass, f"homeassistant/{domain}/{unique_id}/config", data)
     await hass.async_block_till_done()
     assert not ent_registry.async_get_entity_id(domain, mqtt.DOMAIN, unique_id)
+
+
+async def help_test_publishing_with_custom_encoding(
+    hass,
+    mqtt_mock,
+    caplog,
+    domain,
+    config,
+    service,
+    topic,
+    parameters,
+    payload,
+    template,
+    tpl_par="value",
+    tpl_output=None,
+):
+    """Test a service with publishing MQTT payload with different encoding."""
+    # prepare config for tests
+    test_config = {
+        "test1": {"encoding": None, "cmd_tpl": False},
+        "test2": {"encoding": "utf-16", "cmd_tpl": False},
+        "test3": {"encoding": "", "cmd_tpl": False},
+        "test4": {"encoding": "invalid", "cmd_tpl": False},
+        "test5": {"encoding": "", "cmd_tpl": True},
+    }
+    setup_config = []
+    service_data = {}
+    for test_id, test_data in test_config.items():
+        test_config_setup = copy.deepcopy(config)
+        test_config_setup.update(
+            {
+                topic: f"cmd/{test_id}",
+                "name": f"{test_id}",
+            }
+        )
+        if test_data["encoding"] is not None:
+            test_config_setup["encoding"] = test_data["encoding"]
+        if test_data["cmd_tpl"]:
+            test_config_setup[
+                template
+            ] = f"{{{{ (('%.1f'|format({tpl_par}))[0] if is_number({tpl_par}) else {tpl_par}[0]) | ord | pack('b') }}}}"
+        setup_config.append(test_config_setup)
+
+        # setup service data
+        service_data[test_id] = {ATTR_ENTITY_ID: f"{domain}.{test_id}"}
+        if parameters:
+            service_data[test_id].update(parameters)
+
+    # setup test entities
+    assert await async_setup_component(
+        hass,
+        domain,
+        {domain: setup_config},
+    )
+    await hass.async_block_till_done()
+
+    # 1) test with default encoding
+    await hass.services.async_call(
+        domain,
+        service,
+        service_data["test1"],
+        blocking=True,
+    )
+
+    mqtt_mock.async_publish.assert_any_call("cmd/test1", str(payload), 0, False)
+    mqtt_mock.async_publish.reset_mock()
+
+    # 2) test with utf-16 encoding
+    await hass.services.async_call(
+        domain,
+        service,
+        service_data["test2"],
+        blocking=True,
+    )
+    mqtt_mock.async_publish.assert_any_call(
+        "cmd/test2", str(payload).encode("utf-16"), 0, False
+    )
+    mqtt_mock.async_publish.reset_mock()
+
+    # 3) test with no encoding set should fail if payload is a string
+    await hass.services.async_call(
+        domain,
+        service,
+        service_data["test3"],
+        blocking=True,
+    )
+    assert (
+        f"Can't pass-through payload for publishing {payload} on cmd/test3 with no encoding set, need 'bytes'"
+        in caplog.text
+    )
+
+    # 4) test with invalid encoding set should fail
+    await hass.services.async_call(
+        domain,
+        service,
+        service_data["test4"],
+        blocking=True,
+    )
+    assert (
+        f"Can't encode payload for publishing {payload} on cmd/test4 with encoding invalid"
+        in caplog.text
+    )
+
+    # 5) test with command template and raw encoding if specified
+    if not template:
+        return
+
+    await hass.services.async_call(
+        domain,
+        service,
+        service_data["test5"],
+        blocking=True,
+    )
+    mqtt_mock.async_publish.assert_any_call(
+        "cmd/test5", tpl_output or str(payload)[0].encode("utf-8"), 0, False
+    )
+    mqtt_mock.async_publish.reset_mock()