Repository: onicagroup/runway

Issue:
use pydantic to parse arguments for `runway.cfngin.hooks.command`

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/onicagroup/runway/pull/1249?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) Report
> Merging [#1249](https://codecov.io/gh/onicagroup/runway/pull/1249?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) (56369b3) into [master](https://codecov.io/gh/onicagroup/runway/commit/7070937a9d02b4e4e92435e8f28962196627d0a9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) (7070937) will **decrease** coverage by `0.00%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/onicagroup/runway/pull/1249/graphs/tree.svg?width=650&height=150&src=pr&token=Ku28I0RY80&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup)](https://codecov.io/gh/onicagroup/runway/pull/1249?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup)

```diff
@@            Coverage Diff             @@
##           master    #1249      +/-   ##
==========================================
- Coverage   89.67%   89.67%   -0.01%     
==========================================
  Files         247      247              
  Lines       15231    15240       +9     
==========================================
+ Hits        13659    13666       +7     
- Misses       1572     1574       +2     
```


| [Impacted Files](https://codecov.io/gh/onicagroup/runway/pull/1249?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup) | Coverage Δ | |
|---|---|---|
| [runway/cfngin/hooks/command.py](https://codecov.io/gh/onicagroup/runway/pull/1249/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9jb21tYW5kLnB5) | `98.11% <100.00%> (+7.20%)` | :arrow_up: |
| [runway/cfngin/hooks/aws\_lambda.py](https://codecov.io/gh/onicagroup/runway/pull/1249/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9ob29rcy9hd3NfbGFtYmRhLnB5) | `84.10% <0.00%> (-1.66%)` | :arrow_down: |
| [runway/env\_mgr/kbenv.py](https://codecov.io/gh/onicagroup/runway/pull/1249/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Vudl9tZ3Iva2JlbnYucHk=) | `78.76% <0.00%> (-0.69%)` | :arrow_down: |
| [runway/env\_mgr/tfenv.py](https://codecov.io/gh/onicagroup/runway/pull/1249/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Vudl9tZ3IvdGZlbnYucHk=) | `96.46% <0.00%> (-0.51%)` | :arrow_down: |
| [runway/cfngin/exceptions.py](https://codecov.io/gh/onicagroup/runway/pull/1249/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup#diff-cnVud2F5L2Nmbmdpbi9leGNlcHRpb25zLnB5) | `97.15% <0.00%> (+0.94%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/onicagroup/runway/pull/1249?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/onicagroup/runway/pull/1249?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup). Last update [7070937...56369b3](https://codecov.io/gh/onicagroup/runway/pull/1249?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=onicagroup).


Patch:
diff --git a/runway/cfngin/hooks/command.py b/runway/cfngin/hooks/command.py
index 519294f1..f267e534 100644
--- a/runway/cfngin/hooks/command.py
+++ b/runway/cfngin/hooks/command.py
@@ -1,16 +1,51 @@
 """Command hook."""
 import logging
 import os
-from subprocess import PIPE, Popen
+import subprocess
 from typing import Any, Dict, List, Optional, Union
 
 from typing_extensions import TypedDict
 
+from ...utils import BaseModel
 from ..exceptions import ImproperlyConfigured
 
 LOGGER = logging.getLogger(__name__)
 
 
+class RunCommandHookArgs(BaseModel):
+    """Hook arguments for ``run_command``."""
+
+    capture: bool = False
+    """If enabled, capture the command's stdout and stderr, and return them in the hook result."""
+
+    command: Union[str, List[str]]
+    """Command(s) to run."""
+
+    env: Optional[Dict[str, str]] = None
+    """Dictionary of environment variable overrides for the command context.
+    Will be merged with the current environment.
+
+    """
+
+    ignore_status: bool = False
+    """Don't fail the hook if the command returns a non-zero status."""
+
+    interactive: bool = False
+    """If enabled, allow the command to interact with stdin.
+    Otherwise, stdin will be set to the null device.
+
+    """
+
+    quiet: bool = False
+    """Redirect the command's stdout and stderr to the null device, silencing all output.
+    Should not be enabled if ``capture`` is also enabled.
+
+    """
+
+    stdin: Optional[str] = None
+    """String to send to the stdin of the command. Implicitly disables ``interactive``."""
+
+
 class RunCommandResponseTypeDef(TypedDict, total=False):
     """Response from run_command."""
 
@@ -19,36 +54,10 @@ class RunCommandResponseTypeDef(TypedDict, total=False):
     stdout: str
 
 
-def run_command(
-    *,
-    command: Union[str, List[str]],
-    capture: bool = False,
-    interactive: bool = False,
-    ignore_status: bool = False,
-    quiet: bool = False,
-    stdin: Optional[str] = None,
-    env: Optional[Dict[str, str]] = None,
-    **kwargs: Any,
-) -> RunCommandResponseTypeDef:
+def run_command(*__args: Any, **kwargs: Any) -> RunCommandResponseTypeDef:
     """Run a custom command as a hook.
 
-    Args:
-        command: Command(s) to run.
-        capture: If enabled, capture the command's stdout and stderr,
-            and return them in the hook result.
-        interactive: If enabled, allow the command to interact with
-            stdin. Otherwise, stdin will be set to the null device.
-        ignore_status: Don't fail the hook if the command returns a
-            non-zero status.
-        quiet: Redirect the command's stdout and stderr to the null device,
-            silencing all output. Should not be enabled if ``capture`` is also
-            enabled.
-        stdin: String to send to the stdin of the command. Implicitly disables
-            ``interactive``.
-        env: Dictionary of environment variable overrides for the command context.
-            Will be merged with the current environment.
-
-    Additional keyword arguments passed to the function will be forwarded to the
+    Arguments not parsed by the data model will be forwarded to the
     ``subprocess.Popen`` function. Interesting ones include: ``cwd`` and ``shell``.
 
     Examples:
@@ -80,61 +89,67 @@ def run_command(
                     shell: true
 
     """
+    args = RunCommandHookArgs.parse_obj(kwargs)
+
+    # remove parsed args from kwargs
+    for field in RunCommandHookArgs.__fields__:
+        kwargs.pop(field, None)
+
     # remove unneeded args from kwargs
     kwargs.pop("context", None)
     kwargs.pop("provider", None)
 
-    if quiet and capture:
+    if args.quiet and args.capture:
         raise ImproperlyConfigured(
             __name__ + ".run_command",
             ValueError("Cannot enable `quiet` and `capture` options simultaneously"),
         )
 
     with open(os.devnull, "wb") as devnull:
-        if quiet:
+        if args.quiet:
             out_err_type = devnull
-        elif capture:
-            out_err_type = PIPE
+        elif args.capture:
+            out_err_type = subprocess.PIPE
         else:
             out_err_type = None
 
-        if interactive:
+        if args.interactive:
             in_type = None
-        elif stdin:
-            in_type = PIPE
+        elif args.stdin:
+            in_type = subprocess.PIPE
         else:
             in_type = devnull
 
-        if env:
+        if args.env:
             full_env = os.environ.copy()
-            full_env.update(env)
-            env = full_env
+            full_env.update(args.env)
+            args.env = full_env
 
-        LOGGER.info("running command: %s", command)
+        LOGGER.info("running command: %s", args.command)
 
-        with Popen(
-            command,
+        with subprocess.Popen(
+            args.command,
             stdin=in_type,
             stdout=out_err_type,
             stderr=out_err_type,
-            env=env,
+            env=args.env,
             **kwargs,
         ) as proc:
             try:
-                out, err = proc.communicate(stdin)
+                out, err = proc.communicate(args.stdin)
                 status = proc.wait()
 
-                if status == 0 or ignore_status:
+                if status == 0 or args.ignore_status:
                     return {"returncode": proc.returncode, "stdout": out, "stderr": err}
 
                 # Don't print the command line again if we already did earlier
-                if LOGGER.isEnabledFor(logging.INFO):
+                if LOGGER.isEnabledFor(logging.INFO):  # cov: ignore
                     LOGGER.warning("command failed with returncode %d", status)
                 else:
                     LOGGER.warning(
-                        "command failed with returncode %d: %s", status, command
+                        "command failed with returncode %d: %s", status, args.command
                     )
 
                 return {}
-            except Exception:  # pylint: disable=broad-except
+            except Exception:  # pylint: disable=broad-except  # cov: ignore
                 return {}


Test patch:
diff --git a/tests/unit/cfngin/hooks/test_command.py b/tests/unit/cfngin/hooks/test_command.py
index 23f7459a..26fa9b3a 100644
--- a/tests/unit/cfngin/hooks/test_command.py
+++ b/tests/unit/cfngin/hooks/test_command.py
@@ -1,212 +1,98 @@
 """Tests for runway.cfngin.hooks.command."""
-from __future__ import annotations
-
 # pylint: disable=no-self-use
 # pyright: basic
-import os
-import unittest
-from subprocess import PIPE
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    ContextManager,
-    List,
-    Optional,
-    Tuple,
-    Type,
-    Union,
-)
-
-import mock
+from __future__ import annotations
 
-from runway.cfngin.hooks.command import run_command
-from runway.config import CfnginConfig
-from runway.context import CfnginContext
+from typing import TYPE_CHECKING
+
+import pytest
 
-from ..factories import mock_provider
+from runway.cfngin.exceptions import ImproperlyConfigured
+from runway.cfngin.hooks.command import run_command
 
 if TYPE_CHECKING:
-    from types import TracebackType
-
-
-class MockProcess(ContextManager["MockProcess"]):
-    """Mock process."""
-
-    def __init__(
-        self,
-        returncode: int = 0,
-        stdout: Optional[str] = "",
-        stderr: Optional[str] = "",
-    ) -> None:
-        """Instantiate class."""
-        self.returncode = returncode
-        self.stdout = stdout
-        self.stderr = stderr
-        self.stdin = None
-
-    def communicate(self, stdin: str) -> Tuple[Optional[str], Optional[str]]:
-        """Communicate with process."""
-        self.stdin = stdin
-        return (self.stdout, self.stderr)
-
-    def wait(self) -> int:
-        """Wait for process."""
-        return self.returncode
-
-    def kill(self) -> None:
-        """Kill process."""
-        return
-
-    def __enter__(self) -> MockProcess:
-        """Enter the context manager."""
-        return self
-
-    def __exit__(
-        self,
-        exc_type: Optional[Type[BaseException]],
-        exc_value: Optional[BaseException],
-        traceback: Optional[TracebackType],
-    ) -> None:
-        """Exit the context manager."""
-
-
-class TestCommandHook(unittest.TestCase):
-    """Tests for runway.cfngin.hooks.command."""
-
-    def setUp(self) -> None:
-        """Run before tests."""
-        self.context = CfnginContext(
-            config=CfnginConfig.parse_obj(
-                {"namespace": "test", "cfngin_bucket": "test"}
-            )
-        )
-        self.provider = mock_provider(region="us-east-1")
-
-        self.mock_process = MockProcess()
-        self.popen_mock = mock.patch(
-            "runway.cfngin.hooks.command.Popen", return_value=self.mock_process
-        ).start()
-
-    def tearDown(self) -> None:
-        """Run after tests."""
-        self.popen_mock.stop()
-
-    def run_hook(self, *, command: Union[str, List[str]], **kwargs: Any) -> Any:
-        """Run hook."""
-        real_kwargs = {
-            "context": self.context,
-        }
-        real_kwargs.update(kwargs)
-        return run_command(command=command, **real_kwargs)
-
-    def test_command_ok(self) -> None:
-        """Test command ok."""
-        self.mock_process.returncode = 0
-        self.mock_process.stdout = None
-        self.mock_process.stderr = None
-
-        results = self.run_hook(command=["foo"])
-
-        self.assertEqual(results, {"returncode": 0, "stdout": None, "stderr": None})
-        self.popen_mock.assert_called_once_with(
-            ["foo"], stdin=mock.ANY, stdout=None, stderr=None, env=None
-        )
-
-    def test_command_fail(self) -> None:
-        """Test command fail."""
-        self.mock_process.returncode = 1
-        self.mock_process.stdout = None
-        self.mock_process.stderr = None
-
-        results = self.run_hook(command=["foo"])
-
-        self.assertEqual(results, {})
-        self.popen_mock.assert_called_once_with(
-            ["foo"], stdin=mock.ANY, stdout=None, stderr=None, env=None
-        )
-
-    def test_command_ignore_status(self) -> None:
-        """Test command ignore status."""
-        self.mock_process.returncode = 1
-        self.mock_process.stdout = None
-        self.mock_process.stderr = None
-
-        results = self.run_hook(command=["foo"], ignore_status=True)
-
-        self.assertEqual(results, {"returncode": 1, "stdout": None, "stderr": None})
-        self.popen_mock.assert_called_once_with(
-            ["foo"], stdin=mock.ANY, stdout=None, stderr=None, env=None
-        )
-
-    def test_command_quiet(self) -> None:
-        """Test command quiet."""
-        self.mock_process.returncode = 0
-        self.mock_process.stdout = None
-        self.mock_process.stderr = None
-
-        results = self.run_hook(command=["foo"], quiet=True)
-        self.assertEqual(results, {"returncode": 0, "stdout": None, "stderr": None})
-
-        self.popen_mock.assert_called_once_with(
-            ["foo"], stdin=mock.ANY, stdout=mock.ANY, stderr=mock.ANY, env=None
-        )
-
-    def test_command_interactive(self) -> None:
-        """Test command interactive."""
-        self.mock_process.returncode = 0
-        self.mock_process.stdout = None
-        self.mock_process.stderr = None
-
-        results = self.run_hook(command=["foo"], interactive=True)
-        self.assertEqual(results, {"returncode": 0, "stdout": None, "stderr": None})
-
-        self.popen_mock.assert_called_once_with(
-            ["foo"], stdin=None, stdout=None, stderr=None, env=None
-        )
-
-    def test_command_input(self) -> None:
-        """Test command input."""
-        self.mock_process.returncode = 0
-        self.mock_process.stdout = None
-        self.mock_process.stderr = None
-
-        results = self.run_hook(command=["foo"], stdin="hello world")
-        self.assertEqual(results, {"returncode": 0, "stdout": None, "stderr": None})
-
-        self.popen_mock.assert_called_once_with(
-            ["foo"], stdin=PIPE, stdout=None, stderr=None, env=None
-        )
-        self.assertEqual(self.mock_process.stdin, "hello world")
-
-    def test_command_capture(self) -> None:
-        """Test command capture."""
-        self.mock_process.returncode = 0
-        self.mock_process.stdout = "hello"
-        self.mock_process.stderr = "world"
-
-        results = self.run_hook(command=["foo"], capture=True)
-        self.assertEqual(
-            results, {"returncode": 0, "stdout": "hello", "stderr": "world"}
-        )
-
-        self.popen_mock.assert_called_once_with(
-            ["foo"], stdin=mock.ANY, stdout=PIPE, stderr=PIPE, env=None
-        )
-
-    def test_command_env(self) -> None:
-        """Test command env."""
-        self.mock_process.returncode = 0
-        self.mock_process.stdout = None
-        self.mock_process.stderr = None
-
-        with mock.patch.dict(os.environ, {"FOO": "bar"}, clear=True):
-            results = self.run_hook(command=["foo"], env={"hello": "world"})
-
-            self.assertEqual(results, {"returncode": 0, "stdout": None, "stderr": None})
-            self.popen_mock.assert_called_once_with(
-                ["foo"],
-                stdin=mock.ANY,
-                stdout=None,
-                stderr=None,
-                env={"hello": "world", "FOO": "bar"},
-            )
+    from pytest_subprocess import FakeProcess
+
+
+def test_run_command(fake_process: FakeProcess) -> None:
+    """Test run_command."""
+    fake_process.register_subprocess(["foo"], returncode=0)
+    assert run_command(command=["foo"]) == {
+        "returncode": 0,
+        "stderr": None,
+        "stdout": None,
+    }
+
+
+def test_run_command_capture(fake_process: FakeProcess) -> None:
+    """Test run_command with ``capture``."""
+    fake_process.register_subprocess(
+        ["foo"], returncode=0, stderr="bar", stdout="foobar"
+    )
+    assert run_command(command=["foo"], capture=True) == {
+        "returncode": 0,
+        "stderr": b"bar",  # for some reason, pytest-subprocess returns these as bytes
+        "stdout": b"foobar",
+    }
+
+
+def test_run_command_env(fake_process: FakeProcess) -> None:
+    """Test run_command with ``env``."""
+    fake_process.register_subprocess(["foo"], returncode=0)
+    assert run_command(command=["foo"], env={"TEST": "bar"}) == {
+        "returncode": 0,
+        "stderr": None,
+        "stdout": None,
+    }
+
+
+def test_run_command_fail(fake_process: FakeProcess) -> None:
+    """Test run_command non-zero exit code."""
+    fake_process.register_subprocess(["foo"], returncode=1)
+    assert not run_command(command=["foo"])
+
+
+def test_run_command_interactive(fake_process: FakeProcess) -> None:
+    """Test run_command with ``interactive``."""
+    fake_process.register_subprocess(["foo"], returncode=0)
+    assert run_command(command=["foo"], interactive=True) == {
+        "returncode": 0,
+        "stderr": None,
+        "stdout": None,
+    }
+
+
+def test_run_command_ignore_status(fake_process: FakeProcess) -> None:
+    """Test run_command with ``ignore_status``."""
+    fake_process.register_subprocess(["foo"], returncode=1)
+    assert run_command(command=["foo"], ignore_status=True) == {
+        "returncode": 1,
+        "stderr": None,
+        "stdout": None,
+    }
+
+
+def test_run_command_quiet(fake_process: FakeProcess) -> None:
+    """Test run_command with ``quiet``."""
+    fake_process.register_subprocess(["foo"], returncode=0, stderr="", stdout="")
+    assert run_command(command=["foo"], quiet=True) == {
+        "returncode": 0,
+        "stderr": None,
+        "stdout": None,
+    }
+
+
+def test_run_command_raise_improperly_configured() -> None:
+    """Test run_command raise ``ImproperlyConfigured``."""
+    with pytest.raises(ImproperlyConfigured):
+        run_command(command=["foo"], capture=True, quiet=True)
+
+
+def test_run_command_stdin(fake_process: FakeProcess) -> None:
+    """Test run_command with ``stdin``."""
+    fake_process.register_subprocess(["foo"], returncode=0)
+    assert run_command(command=["foo"], stdin="bar") == {
+        "returncode": 0,
+        "stderr": None,
+        "stdout": None,
+    }