xref: /aosp_15_r20/external/pigweed/pw_cli/py/tool_runner_test.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"""Tests for pw_cli.tool_runner."""
15
16import subprocess
17import sys
18import unittest
19from typing import Any, Iterable
20
21from pw_cli import tool_runner
22from pw_cli.tool_runner import ToolRunner
23
24
25class TestToolRunner(unittest.TestCase):
26    """Tests for tool_runner.TestToolRunner."""
27
28    def test_basic_subprocess_runner(self):
29        runner = tool_runner.BasicSubprocessRunner()
30        if sys.platform == 'win32':
31            result = runner('python.exe', ('-c', 'print("hello world")'))
32        else:
33            result = runner('echo', ('hello', 'world'))
34        self.assertEqual(result.returncode, 0)
35        self.assertIn('hello world', result.stdout.decode())
36
37
38class FakeTool(ToolRunner):
39    def __init__(self) -> None:
40        self.received_args: list[str] = []
41        self.received_kwargs: dict[str, Any] = {}
42
43    def _run_tool(
44        self, tool: str, args, **kwargs
45    ) -> subprocess.CompletedProcess:
46        self.received_args = list(args)
47        self.received_kwargs = kwargs
48
49        full_command = ' '.join((tool, *tuple(args)))
50
51        return subprocess.CompletedProcess(
52            args=full_command,
53            returncode=0xFF,
54            stderr=f'I do not know how to `{full_command}`'.encode(),
55            stdout=b'Failed to execute command',
56        )
57
58
59class FakeToolWithCustomArgs(FakeTool):
60    @staticmethod
61    def _custom_args() -> Iterable[str]:
62        return ['pw_custom_arg', 'pw_2_custom_2_arg']
63
64
65class ToolRunnerCallTest(unittest.TestCase):
66    """Tests argument forwarding to ToolRunner implementations."""
67
68    def test_fake_tool_without_custom_args(self):
69        tool = FakeTool()
70
71        tool(
72            'rm',
73            ('-rf', '/'),
74            capture_output=True,
75            pw_custom_arg='should not be forwarded',
76        )
77
78        self.assertEqual(tool.received_args, ['-rf', '/'])
79        self.assertEqual(
80            tool.received_kwargs,
81            {
82                'capture_output': True,
83                'stdout': subprocess.PIPE,
84                'stderr': subprocess.PIPE,
85            },
86        )
87
88    def test_fake_tool_with_custom_args(self):
89        tool = FakeToolWithCustomArgs()
90
91        tool(
92            'rm',
93            ('-rf', '/'),
94            capture_output=True,
95            pw_custom_arg='should be forwarded',
96            pw_2_custom_2_arg='this one too',
97            pw_foo='but not this',
98        )
99
100        self.assertEqual(tool.received_args, ['-rf', '/'])
101        self.assertEqual(
102            tool.received_kwargs,
103            {
104                'capture_output': True,
105                'stdout': subprocess.PIPE,
106                'stderr': subprocess.PIPE,
107                'pw_custom_arg': 'should be forwarded',
108                'pw_2_custom_2_arg': 'this one too',
109            },
110        )
111
112
113if __name__ == '__main__':
114    unittest.main()
115