Merge pull request #1220 from stefan-jonasson/command_line.py

New notification platform "command line"

homeassistant/components/notify/command_line.py

@@ -0,0 +1,48 @@
+"""
+homeassistant.components.notify.command_line
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+command_line notification service.
+
+For more details about this platform, please refer to the documentation at
+https://home-assistant.io/components/notify.command_line/
+"""
+import logging
+import subprocess
+from homeassistant.helpers import validate_config
+from homeassistant.components.notify import (
+    DOMAIN, BaseNotificationService)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def get_service(hass, config):
+    """ Get the Command Line notification service. """
+
+    if not validate_config({DOMAIN: config},
+                           {DOMAIN: ['command']},
+                           _LOGGER):
+        return None
+
+    command = config['command']
+
+    return CommandLineNotificationService(command)
+
+
+# pylint: disable=too-few-public-methods
+class CommandLineNotificationService(BaseNotificationService):
+    """ Implements notification service for the Command Line service. """
+
+    def __init__(self, command):
+        self.command = command
+
+    def send_message(self, message="", **kwargs):
+        """ Send a message to a command_line. """
+
+        try:
+            proc = subprocess.Popen(self.command, universal_newlines=True,
+                                    stdin=subprocess.PIPE, shell=True)
+            proc.communicate(input=message)
+            if proc.returncode != 0:
+                _LOGGER.error('Command failed: %s', self.command)
+        except subprocess.SubprocessError:
+            _LOGGER.error('Error trying to exec Command: %s', self.command)

tests/components/notify/test_command_line.py

@@ -0,0 +1,58 @@
+"""
+tests.components.notify.test_command_line
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Tests command line notification.
+"""
+import os
+import tempfile
+import unittest
+
+from homeassistant import core
+import homeassistant.components.notify as notify
+from unittest.mock import patch
+
+
+class TestCommandLine(unittest.TestCase):
+    """ Test the command line. """
+
+    def setUp(self):  # pylint: disable=invalid-name
+        self.hass = core.HomeAssistant()
+
+    def tearDown(self):  # pylint: disable=invalid-name
+        """ Stop down stuff we started. """
+        self.hass.stop()
+
+    def test_command_line_output(self):
+        with tempfile.TemporaryDirectory() as tempdirname:
+            filename = os.path.join(tempdirname, 'message.txt')
+            message = 'one, two, testing, testing'
+            self.assertTrue(notify.setup(self.hass, {
+                'notify': {
+                    'name': 'test',
+                    'platform': 'command_line',
+                    'command': 'echo $(cat) > {}'.format(filename)
+                }
+            }))
+
+            self.hass.services.call('notify', 'test', {'message': message},
+                                    blocking=True)
+
+            result = open(filename).read()
+            # the echo command adds a line break
+            self.assertEqual(result, "{}\n".format(message))
+
+    @patch('homeassistant.components.notify.command_line._LOGGER.error')
+    def test_error_for_none_zero_exit_code(self, mock_error):
+        """ Test if an error if logged for non zero exit codes. """
+        self.assertTrue(notify.setup(self.hass, {
+            'notify': {
+                'name': 'test',
+                'platform': 'command_line',
+                'command': 'echo $(cat); exit 1'
+            }
+        }))
+
+        self.hass.services.call('notify', 'test', {'message': 'error'},
+                                blocking=True)
+        self.assertEqual(1, mock_error.call_count)