Support templating MQTT triggers (#45614)

* Add support for limited templates (no HASS access)

* Pass variables to automation triggers

* Support templates in MQTT triggers

* Spelling

* Handle trigger referenced by variables

* Raise on unsupported function in limited templates

* Validate MQTT trigger schema in MQTT device trigger

* Add trigger_variables to automation config schema

* Don't print stacktrace when setting up trigger throws

* Make pylint happy

* Add trigger_variables to variables

* Add debug prints, document limited template

* Add tests

* Validate MQTT trigger topic early when possible

* Improve valid_subscribe_topic_template

tests/components/mqtt/test_trigger.py

@@ -81,6 +81,58 @@ async def test_if_fires_on_topic_and_payload_match(hass, calls):
     assert len(calls) == 1
 
 
+async def test_if_fires_on_templated_topic_and_payload_match(hass, calls):
+    """Test if message is fired on templated topic and payload match."""
+    assert await async_setup_component(
+        hass,
+        automation.DOMAIN,
+        {
+            automation.DOMAIN: {
+                "trigger": {
+                    "platform": "mqtt",
+                    "topic": "test-topic-{{ sqrt(16)|round }}",
+                    "payload": '{{ "foo"|regex_replace("foo", "bar") }}',
+                },
+                "action": {"service": "test.automation"},
+            }
+        },
+    )
+
+    async_fire_mqtt_message(hass, "test-topic-", "foo")
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    async_fire_mqtt_message(hass, "test-topic-4", "foo")
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    async_fire_mqtt_message(hass, "test-topic-4", "bar")
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+
+
+async def test_non_allowed_templates(hass, calls, caplog):
+    """Test non allowed function in template."""
+    assert await async_setup_component(
+        hass,
+        automation.DOMAIN,
+        {
+            automation.DOMAIN: {
+                "trigger": {
+                    "platform": "mqtt",
+                    "topic": "test-topic-{{ states() }}",
+                },
+                "action": {"service": "test.automation"},
+            }
+        },
+    )
+
+    assert (
+        "Got error 'TemplateError: str: Use of 'states' is not supported in limited templates' when setting up triggers"
+        in caplog.text
+    )
+
+
 async def test_if_not_fires_on_topic_but_no_payload_match(hass, calls):
     """Test if message is not fired on topic but no payload."""
     assert await async_setup_component(