xref: /aosp_15_r20/external/pigweed/pw_cli/py/pw_cli/tool_runner.py (revision 61c4878ac05f98d0ceed94b57d316916de578985) 
1# Copyright 2024 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""A subprocess wrapper that enables injection of externally-provided tools."""
15
16import abc
17from pathlib import Path
18from typing import Iterable
19
20import subprocess
21
22
23class ToolRunner(abc.ABC):
24    """A callable interface that runs the requested tool as a subprocess.
25
26    This class is used to support subprocess-like semantics while allowing
27    injection of wrappers that enable testing, finer granularity identifying
28    where tools fail, and stricter control of which binaries are called.
29
30    By default, all subprocess output is captured.
31    """
32
33    def __call__(
34        self,
35        tool: str,
36        args: Iterable[str | Path],
37        stdout: int | None = subprocess.PIPE,
38        stderr: int | None = subprocess.PIPE,
39        **kwargs,
40    ) -> subprocess.CompletedProcess:
41        """Calls ``tool`` with the provided ``args``.
42
43        ``**kwargs`` are forwarded to the underlying ``subprocess.run()``
44        for the requested tool.
45
46        By default, all subprocess output is captured.
47
48        Returns:
49            The ``subprocess.CompletedProcess`` result of running the requested
50            tool.
51        """
52        additional_kwargs = set(self._custom_args())
53        allowed_kwargs = {
54            key: value
55            for key, value in kwargs.items()
56            if not key.startswith('pw_') or key in additional_kwargs
57        }
58
59        return self._run_tool(
60            tool,
61            args,
62            stderr=stderr,
63            stdout=stdout,
64            **allowed_kwargs,
65        )
66
67    @staticmethod
68    def _custom_args() -> Iterable[str]:
69        """List of additional keyword arguments accepted by this tool.
70
71        By default, all kwargs passed into a tool are forwarded to
72        ``subprocess.run()``. However, some tools have extra arguments custom
73        to them, which are not valid for ``subprocess.run()``. Tools requiring
74        these custom args should override this method, listing the arguments
75        they accept.
76
77        To make filtering custom arguments possible, they must be prefixed
78        with  ``pw_``.
79        """
80        return []
81
82    @abc.abstractmethod
83    def _run_tool(
84        self, tool: str, args, **kwargs
85    ) -> subprocess.CompletedProcess:
86        """Implements the subprocess runner logic.
87
88        Calls ``tool`` with the provided ``args``. ``**kwargs`` not listed in
89        ``_custom_args`` are forwarded to the underlying ``subprocess.run()``
90        for the requested tool.
91
92        Returns:
93            The ``subprocess.CompletedProcess`` result of running the requested
94            tool.
95        """
96
97
98class BasicSubprocessRunner(ToolRunner):
99    """A simple ToolRunner that calls subprocess.run()."""
100
101    def _run_tool(
102        self, tool: str, args, **kwargs
103    ) -> subprocess.CompletedProcess:
104        return subprocess.run([tool, *args], **kwargs)
105