Merge pull request #513 from balloob/component-command

Add shell_command component

homeassistant/components/shell_command.py

@@ -0,0 +1,48 @@
+"""
+homeassistant.components.shell_command
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Component to expose shell commands as services.
+
+shell_command:
+  restart_pow: touch ~/.pow/restart.txt
+
+"""
+import logging
+import subprocess
+
+from homeassistant.util import slugify
+
+DOMAIN = 'shell_command'
+DEPENDENCIES = []
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def setup(hass, config):
+    """ Sets up the shell_command component. """
+    conf = config.get(DOMAIN)
+
+    if not isinstance(conf, dict):
+        _LOGGER.error('Expected configuration to be a dictionary')
+        return False
+
+    for name in conf.keys():
+        if name != slugify(name):
+            _LOGGER.error('Invalid service name: %s. Try %s',
+                          name, slugify(name))
+            return False
+
+    def service_handler(call):
+        """ Execute a shell command service. """
+        try:
+            subprocess.call(conf[call.service].split(' '),
+                            stdout=subprocess.DEVNULL,
+                            stderr=subprocess.DEVNULL)
+        except subprocess.SubprocessError:
+            _LOGGER.exception('Error running command')
+
+    for name in conf.keys():
+        hass.services.register(DOMAIN, name, service_handler)
+
+    return True

tests/components/test_shell_command.py

@@ -0,0 +1,71 @@
+"""
+tests.test_shell_command
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Tests demo component.
+"""
+import os
+import tempfile
+import unittest
+from unittest.mock import patch
+from subprocess import SubprocessError
+
+from homeassistant import core
+from homeassistant.components import shell_command
+
+
+class TestShellCommand(unittest.TestCase):
+    """ Test the demo module. """
+
+    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_executing_service(self):
+        """ Test if able to call a configured service. """
+        with tempfile.TemporaryDirectory() as tempdirname:
+            path = os.path.join(tempdirname, 'called.txt')
+            self.assertTrue(shell_command.setup(self.hass, {
+                'shell_command': {
+                    'test_service': "touch {}".format(path)
+                }
+            }))
+
+            self.hass.services.call('shell_command', 'test_service',
+                                    blocking=True)
+
+            self.assertTrue(os.path.isfile(path))
+
+    def test_config_not_dict(self):
+        """ Test if config is not a dict. """
+        self.assertFalse(shell_command.setup(self.hass, {
+            'shell_command': ['some', 'weird', 'list']
+            }))
+
+    def test_config_not_valid_service_names(self):
+        """ Test if config contains invalid service names. """
+        self.assertFalse(shell_command.setup(self.hass, {
+            'shell_command': {
+                'this is invalid because space': 'touch bla.txt'
+            }}))
+
+    @patch('homeassistant.components.shell_command.subprocess.call',
+           side_effect=SubprocessError)
+    @patch('homeassistant.components.shell_command._LOGGER.error')
+    def test_subprocess_raising_error(self, mock_call, mock_error):
+        with tempfile.TemporaryDirectory() as tempdirname:
+            path = os.path.join(tempdirname, 'called.txt')
+            self.assertTrue(shell_command.setup(self.hass, {
+                'shell_command': {
+                    'test_service': "touch {}".format(path)
+                }
+            }))
+
+            self.hass.services.call('shell_command', 'test_service',
+                                    blocking=True)
+
+            self.assertFalse(os.path.isfile(path))
+            self.assertEqual(1, mock_error.call_count)