diff --git a/homeassistant/components/notify/command_line.py b/homeassistant/components/notify/command_line.py new file mode 100644 index 00000000000..025046f1e7c --- /dev/null +++ b/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) diff --git a/tests/components/notify/test_command_line.py b/tests/components/notify/test_command_line.py new file mode 100644 index 00000000000..f920590a21c --- /dev/null +++ b/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)