coverage

homeassistant/components/command_line/notify.py

@@ -9,10 +9,12 @@ from typing import Any, cast
 from homeassistant.components.notify import BaseNotificationService
 from homeassistant.const import CONF_COMMAND
 from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import TemplateError
+from homeassistant.helpers.template import Template
 from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 from homeassistant.util.process import kill_subprocess
 
-from .const import CONF_COMMAND_TIMEOUT
+from .const import CONF_COMMAND_TIMEOUT, LOGGER
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -44,8 +46,31 @@ class CommandLineNotificationService(BaseNotificationService):
 
     def send_message(self, message: str = "", **kwargs: Any) -> None:
         """Send a message to a command line."""
+        command = self.command
+        if " " not in command:
+            prog = command
+            args = None
+            args_compiled = None
+        else:
+            prog, args = command.split(" ", 1)
+            args_compiled = Template(args, self.hass)
+
+        rendered_args = None
+        if args_compiled:
+            args_to_render = {"arguments": args}
+            try:
+                rendered_args = args_compiled.async_render(args_to_render)
+            except TemplateError as ex:
+                LOGGER.exception("Error rendering command template: %s", ex)
+                return
+
+        if rendered_args != args:
+            command = f"{prog} {rendered_args}"
+
+        LOGGER.debug("Running command: %s, with message: %s", command, message)
+
         with subprocess.Popen(  # noqa: S602 # shell by design
-            self.command,
+            command,
             universal_newlines=True,
             stdin=subprocess.PIPE,
             close_fds=False,  # required for posix_spawn
@@ -57,10 +82,10 @@ class CommandLineNotificationService(BaseNotificationService):
                     _LOGGER.error(
                         "Command failed (with return code %s): %s",
                         proc.returncode,
-                        self.command,
+                        command,
                     )
             except subprocess.TimeoutExpired:
-                _LOGGER.error("Timeout for command: %s", self.command)
+                _LOGGER.error("Timeout for command: %s", command)
                 kill_subprocess(proc)
             except subprocess.SubprocessError:
-                _LOGGER.error("Error trying to exec command: %s", self.command)
+                _LOGGER.error("Error trying to exec command: %s", command)

tests/components/command_line/test_notify.py

@@ -100,6 +100,100 @@ async def test_command_line_output(hass: HomeAssistant) -> None:
         assert message == await hass.async_add_executor_job(Path(filename).read_text)
 
 
+async def test_command_line_output_single_command(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test the command line output."""
+
+    await setup.async_setup_component(
+        hass,
+        DOMAIN,
+        {
+            "command_line": [
+                {
+                    "notify": {
+                        "command": "echo",
+                        "name": "Test3",
+                    }
+                }
+            ]
+        },
+    )
+    await hass.async_block_till_done()
+
+    assert hass.services.has_service(NOTIFY_DOMAIN, "test3")
+
+    await hass.services.async_call(
+        NOTIFY_DOMAIN, "test3", {"message": "test message"}, blocking=True
+    )
+    assert "Running command: echo, with message: test message" in caplog.text
+
+
+async def test_command_template(hass: HomeAssistant) -> None:
+    """Test the command line output using template as command."""
+
+    with tempfile.TemporaryDirectory() as tempdirname:
+        filename = os.path.join(tempdirname, "message.txt")
+        message = "one, two, testing, testing"
+        hass.states.async_set("sensor.test_state", filename)
+        await setup.async_setup_component(
+            hass,
+            DOMAIN,
+            {
+                "command_line": [
+                    {
+                        "notify": {
+                            "command": "cat > {{ states.sensor.test_state.state }}",
+                            "name": "Test3",
+                        }
+                    }
+                ]
+            },
+        )
+        await hass.async_block_till_done()
+
+        assert hass.services.has_service(NOTIFY_DOMAIN, "test3")
+
+        await hass.services.async_call(
+            NOTIFY_DOMAIN, "test3", {"message": message}, blocking=True
+        )
+        assert message == await hass.async_add_executor_job(Path(filename).read_text)
+
+
+async def test_command_incorrect_template(
+    hass: HomeAssistant, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Test the command line output using template as command which isn't working."""
+
+    message = "one, two, testing, testing"
+    await setup.async_setup_component(
+        hass,
+        DOMAIN,
+        {
+            "command_line": [
+                {
+                    "notify": {
+                        "command": "cat > {{ this template doesn't parse ",
+                        "name": "Test3",
+                    }
+                }
+            ]
+        },
+    )
+    await hass.async_block_till_done()
+
+    assert hass.services.has_service(NOTIFY_DOMAIN, "test3")
+
+    await hass.services.async_call(
+        NOTIFY_DOMAIN, "test3", {"message": message}, blocking=True
+    )
+
+    assert (
+        "Error rendering command template: TemplateSyntaxError: expected token"
+        in caplog.text
+    )
+
+
 @pytest.mark.parametrize(
     "get_config",
     [