diff --git a/tests/components/shell_command/test_init.py b/tests/components/shell_command/test_init.py index 1f8e442e93d..f5ad37cc617 100644 --- a/tests/components/shell_command/test_init.py +++ b/tests/components/shell_command/test_init.py @@ -1,15 +1,13 @@ """The tests for the Shell command component.""" -import asyncio + import os import tempfile from typing import Tuple -import unittest from homeassistant.components import shell_command -from homeassistant.setup import async_setup_component, setup_component +from homeassistant.setup import async_setup_component -from tests.async_mock import Mock, patch -from tests.common import get_test_home_assistant +from tests.async_mock import MagicMock, patch def mock_process_creator(error: bool = False): @@ -22,161 +20,151 @@ def mock_process_creator(error: bool = False): """ return b"I am stdout", b"I am stderr" - mock_process = Mock() + mock_process = MagicMock() mock_process.communicate = communicate mock_process.returncode = int(error) return mock_process -class TestShellCommand(unittest.TestCase): - """Test the shell_command component.""" - - def setUp(self): # pylint: disable=invalid-name - """Set up things to be run when tests are started. - - Also seems to require a child watcher attached to the loop when run - from pytest. - """ - self.hass = get_test_home_assistant() - asyncio.get_child_watcher().attach_loop(self.hass.loop) - self.addCleanup(self.tear_down_cleanup) - - def tear_down_cleanup(self): - """Stop everything that was 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") - assert setup_component( - self.hass, - shell_command.DOMAIN, - {shell_command.DOMAIN: {"test_service": f"date > {path}"}}, - ) - - self.hass.services.call("shell_command", "test_service", blocking=True) - self.hass.block_till_done() - assert os.path.isfile(path) - - def test_config_not_dict(self): - """Test that setup fails if config is not a dict.""" - assert not setup_component( - self.hass, +async def test_executing_service(hass): + """Test if able to call a configured service.""" + with tempfile.TemporaryDirectory() as tempdirname: + path = os.path.join(tempdirname, "called.txt") + assert await async_setup_component( + hass, shell_command.DOMAIN, - {shell_command.DOMAIN: ["some", "weird", "list"]}, + {shell_command.DOMAIN: {"test_service": f"date > {path}"}}, ) + await hass.async_block_till_done() - def test_config_not_valid_service_names(self): - """Test that setup fails if config contains invalid service names.""" - assert not setup_component( - self.hass, - shell_command.DOMAIN, - {shell_command.DOMAIN: {"this is invalid because space": "touch bla.txt"}}, - ) + await hass.services.async_call("shell_command", "test_service", blocking=True) + await hass.async_block_till_done() + assert os.path.isfile(path) - @patch( - "homeassistant.components.shell_command.asyncio.subprocess" - ".create_subprocess_shell" + +async def test_config_not_dict(hass): + """Test that setup fails if config is not a dict.""" + assert not await async_setup_component( + hass, + shell_command.DOMAIN, + {shell_command.DOMAIN: ["some", "weird", "list"]}, ) - def test_template_render_no_template(self, mock_call): - """Ensure shell_commands without templates get rendered properly.""" - mock_call.return_value = mock_process_creator(error=False) - assert setup_component( - self.hass, + +async def test_config_not_valid_service_names(hass): + """Test that setup fails if config contains invalid service names.""" + assert not await async_setup_component( + hass, + shell_command.DOMAIN, + {shell_command.DOMAIN: {"this is invalid because space": "touch bla.txt"}}, + ) + + +@patch( + "homeassistant.components.shell_command.asyncio.subprocess" + ".create_subprocess_shell" +) +async def test_template_render_no_template(mock_call, hass): + """Ensure shell_commands without templates get rendered properly.""" + mock_call.return_value = mock_process_creator(error=False) + + assert await async_setup_component( + hass, + shell_command.DOMAIN, + {shell_command.DOMAIN: {"test_service": "ls /bin"}}, + ) + await hass.async_block_till_done() + + await hass.services.async_call("shell_command", "test_service", blocking=True) + await hass.async_block_till_done() + cmd = mock_call.mock_calls[0][1][0] + + assert mock_call.call_count == 1 + assert "ls /bin" == cmd + + +@patch( + "homeassistant.components.shell_command.asyncio.subprocess" + ".create_subprocess_exec" +) +async def test_template_render(mock_call, hass): + """Ensure shell_commands with templates get rendered properly.""" + hass.states.async_set("sensor.test_state", "Works") + mock_call.return_value = mock_process_creator(error=False) + assert await async_setup_component( + hass, + shell_command.DOMAIN, + { + shell_command.DOMAIN: { + "test_service": ("ls /bin {{ states.sensor.test_state.state }}") + } + }, + ) + + await hass.services.async_call("shell_command", "test_service", blocking=True) + + await hass.async_block_till_done() + cmd = mock_call.mock_calls[0][1] + + assert mock_call.call_count == 1 + assert ("ls", "/bin", "Works") == cmd + + +@patch( + "homeassistant.components.shell_command.asyncio.subprocess" + ".create_subprocess_shell" +) +@patch("homeassistant.components.shell_command._LOGGER.error") +async def test_subprocess_error(mock_error, mock_call, hass): + """Test subprocess that returns an error.""" + mock_call.return_value = mock_process_creator(error=True) + with tempfile.TemporaryDirectory() as tempdirname: + path = os.path.join(tempdirname, "called.txt") + assert await async_setup_component( + hass, shell_command.DOMAIN, - {shell_command.DOMAIN: {"test_service": "ls /bin"}}, + {shell_command.DOMAIN: {"test_service": f"touch {path}"}}, ) - self.hass.services.call("shell_command", "test_service", blocking=True) - - self.hass.block_till_done() - cmd = mock_call.mock_calls[0][1][0] - + await hass.services.async_call("shell_command", "test_service", blocking=True) + await hass.async_block_till_done() assert mock_call.call_count == 1 - assert "ls /bin" == cmd + assert mock_error.call_count == 1 + assert not os.path.isfile(path) - @patch( - "homeassistant.components.shell_command.asyncio.subprocess" - ".create_subprocess_exec" + +@patch("homeassistant.components.shell_command._LOGGER.debug") +async def test_stdout_captured(mock_output, hass): + """Test subprocess that has stdout.""" + test_phrase = "I have output" + assert await async_setup_component( + hass, + shell_command.DOMAIN, + {shell_command.DOMAIN: {"test_service": f"echo {test_phrase}"}}, ) - def test_template_render(self, mock_call): - """Ensure shell_commands with templates get rendered properly.""" - self.hass.states.set("sensor.test_state", "Works") - mock_call.return_value = mock_process_creator(error=False) - assert setup_component( - self.hass, - shell_command.DOMAIN, - { - shell_command.DOMAIN: { - "test_service": ("ls /bin {{ states.sensor.test_state.state }}") - } - }, - ) - self.hass.services.call("shell_command", "test_service", blocking=True) + await hass.services.async_call("shell_command", "test_service", blocking=True) - self.hass.block_till_done() - cmd = mock_call.mock_calls[0][1] + await hass.async_block_till_done() + assert mock_output.call_count == 1 + assert test_phrase.encode() + b"\n" == mock_output.call_args_list[0][0][-1] - assert mock_call.call_count == 1 - assert ("ls", "/bin", "Works") == cmd - @patch( - "homeassistant.components.shell_command.asyncio.subprocess" - ".create_subprocess_shell" +@patch("homeassistant.components.shell_command._LOGGER.debug") +async def test_stderr_captured(mock_output, hass): + """Test subprocess that has stderr.""" + test_phrase = "I have error" + assert await async_setup_component( + hass, + shell_command.DOMAIN, + {shell_command.DOMAIN: {"test_service": f">&2 echo {test_phrase}"}}, ) - @patch("homeassistant.components.shell_command._LOGGER.error") - def test_subprocess_error(self, mock_error, mock_call): - """Test subprocess that returns an error.""" - mock_call.return_value = mock_process_creator(error=True) - with tempfile.TemporaryDirectory() as tempdirname: - path = os.path.join(tempdirname, "called.txt") - assert setup_component( - self.hass, - shell_command.DOMAIN, - {shell_command.DOMAIN: {"test_service": f"touch {path}"}}, - ) - self.hass.services.call("shell_command", "test_service", blocking=True) + await hass.services.async_call("shell_command", "test_service", blocking=True) - self.hass.block_till_done() - assert mock_call.call_count == 1 - assert mock_error.call_count == 1 - assert not os.path.isfile(path) - - @patch("homeassistant.components.shell_command._LOGGER.debug") - def test_stdout_captured(self, mock_output): - """Test subprocess that has stdout.""" - test_phrase = "I have output" - assert setup_component( - self.hass, - shell_command.DOMAIN, - {shell_command.DOMAIN: {"test_service": f"echo {test_phrase}"}}, - ) - - self.hass.services.call("shell_command", "test_service", blocking=True) - - self.hass.block_till_done() - assert mock_output.call_count == 1 - assert test_phrase.encode() + b"\n" == mock_output.call_args_list[0][0][-1] - - @patch("homeassistant.components.shell_command._LOGGER.debug") - def test_stderr_captured(self, mock_output): - """Test subprocess that has stderr.""" - test_phrase = "I have error" - assert setup_component( - self.hass, - shell_command.DOMAIN, - {shell_command.DOMAIN: {"test_service": f">&2 echo {test_phrase}"}}, - ) - - self.hass.services.call("shell_command", "test_service", blocking=True) - - self.hass.block_till_done() - assert mock_output.call_count == 1 - assert test_phrase.encode() + b"\n" == mock_output.call_args_list[0][0][-1] + await hass.async_block_till_done() + assert mock_output.call_count == 1 + assert test_phrase.encode() + b"\n" == mock_output.call_args_list[0][0][-1] async def test_do_no_run_forever(hass, caplog):