generated from salesforce/oss-template
-
Notifications
You must be signed in to change notification settings - Fork 4
Expand file tree
/
Copy pathtest_cmd.py
More file actions
170 lines (135 loc) · 5.46 KB
/
test_cmd.py
File metadata and controls
170 lines (135 loc) · 5.46 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
from __future__ import annotations
import subprocess
from unittest.mock import Mock, patch
import pytest
from datacustomcode.cmd import (
CalledProcessError,
_cmd_output,
_force_bytes,
_oserror_to_output,
cmd_output,
)
class TestCmd:
def test_force_bytes(self):
"""Test _force_bytes conversion of different types to bytes."""
# String
assert _force_bytes("test") == b"test"
# Already bytes
assert _force_bytes(b"test") == b"test"
# Exception with string representation
error = ValueError("error message")
assert _force_bytes(error) == b"error message"
# Object without string representation
class UnprintableObject:
def __str__(self):
raise TypeError()
obj = UnprintableObject()
assert b"unprintable UnprintableObject object" in _force_bytes(obj)
def test_oserror_to_output(self):
"""Test converting OSError to command output format."""
error = OSError("test error")
returncode, stdout, stderr = _oserror_to_output(error)
assert returncode == 1
assert stdout == b"test error\n"
assert stderr is None
def test_called_process_error_formatting(self):
"""Test CalledProcessError string formatting."""
error = CalledProcessError(
returncode=1,
cmd=("ls", "-l"),
stdout=b"stdout content",
stderr=b"stderr content",
)
error_str = str(error)
assert "command: ('ls', '-l')" in error_str
assert "return code: 1" in error_str
assert "stdout:stdout content" in error_str
assert "stderr:\n stderr content" in error_str
# Test with None stderr
error = CalledProcessError(
returncode=1,
cmd=("ls", "-l"),
stdout=b"stdout content",
stderr=None,
)
error_str = str(error)
assert "stderr: (none)" in error_str
@patch("subprocess.Popen")
def test_cmd_output_success(self, mock_popen):
"""Test _cmd_output with successful command execution."""
# Setup mock
process_mock = Mock()
process_mock.communicate.return_value = (b"command output", b"")
process_mock.returncode = 0
mock_popen.return_value = process_mock
# Call function
returncode, stdout, stderr = _cmd_output("ls", "-l")
# Assertions
assert returncode == 0
assert stdout == b"command output"
assert stderr == b""
mock_popen.assert_called_once()
# Check that default kwargs were set
_, kwargs = mock_popen.call_args
assert kwargs["stdin"] == subprocess.PIPE
assert kwargs["stdout"] == subprocess.PIPE
assert kwargs["stderr"] == subprocess.PIPE
@patch("subprocess.Popen")
def test_cmd_output_failure(self, mock_popen):
"""Test _cmd_output with command failure."""
# Setup mock
process_mock = Mock()
process_mock.communicate.return_value = (b"", b"command error")
process_mock.returncode = 1
mock_popen.return_value = process_mock
# Call function with check=True (default)
with pytest.raises(CalledProcessError) as exc_info:
_cmd_output("ls", "-l")
error = exc_info.value
assert error.returncode == 1
assert error.stdout == b""
assert error.stderr == b"command error"
# Call function with check=False
returncode, stdout, stderr = _cmd_output("ls", "-l", check=False)
assert returncode == 1
assert stdout == b""
assert stderr == b"command error"
@patch("subprocess.Popen")
def test_cmd_output_oserror(self, mock_popen):
"""Test _cmd_output when subprocess.Popen raises OSError."""
# Setup mock to raise OSError
mock_popen.side_effect = OSError("command not found")
# Call function with check=True (default)
with pytest.raises(CalledProcessError) as exc_info:
_cmd_output("nonexistent_command")
error = exc_info.value
assert error.returncode == 1
assert b"command not found" in error.stdout
# Call function with check=False
returncode, stdout, stderr = _cmd_output("nonexistent_command", check=False)
assert returncode == 1
assert b"command not found" in stdout
assert stderr is None
@patch("datacustomcode.cmd._cmd_output")
def test_cmd_output_wrapper(self, mock_cmd_output):
"""Test cmd_output wrapper function."""
# Setup mock
mock_cmd_output.return_value = (0, b"command output", b"")
# Call function
result = cmd_output("ls", "-l")
# Assertions
assert result == "command output"
mock_cmd_output.assert_called_once_with("ls", "-l")
# Test with non-zero return code
mock_cmd_output.return_value = (1, b"", b"error")
result = cmd_output("ls", "-l", check=False)
assert result == ""
@patch("datacustomcode.cmd._cmd_output")
def test_cmd_output_with_custom_kwargs(self, mock_cmd_output):
"""Test cmd_output with custom keyword arguments."""
mock_cmd_output.return_value = (0, b"output", None)
cmd_output("ls", "-l", env={"PATH": "/usr/bin"}, cwd="/tmp")
# Verify custom kwargs were passed through
_, kwargs = mock_cmd_output.call_args
assert kwargs["env"] == {"PATH": "/usr/bin"}
assert kwargs["cwd"] == "/tmp"