Skip to content

Commit d368a01

Browse files
committed
test(fsm): Add FSM cmdline-related unit tests.
Unit tests for TheMachine cmdline-related methods in fsm.py. This test module provides comprehensive coverage for the command line retrieval functions that work across different platforms (Windows, Linux, Unix). Tested functions: - _get_cmdline_windows(): Retrieves command line on Windows using ctypes - _get_cmdline_linux_proc(): Retrieves command line from /proc/self/cmdline - _get_cmdline_unix_ps(): Retrieves command line using ps command - _get_cmdline_unix(): Dispatches to appropriate Unix method - _get_cmdline(): Main entry point with platform detection and error handling Signed-off-by: Paulo Vital <[email protected]>
1 parent 86b0585 commit d368a01

File tree

1 file changed

+384
-0
lines changed

1 file changed

+384
-0
lines changed

tests/test_fsm_cmdline.py

Lines changed: 384 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,384 @@
1+
# (c) Copyright IBM Corp. 2025
2+
"""
3+
Unit tests for TheMachine cmdline-related methods in fsm.py.
4+
5+
This test module provides comprehensive coverage for the command line retrieval
6+
functions that work across different platforms (Windows, Linux, Unix).
7+
8+
Tested functions:
9+
- _get_cmdline_windows(): Retrieves command line on Windows using ctypes
10+
- _get_cmdline_linux_proc(): Retrieves command line from /proc/self/cmdline
11+
- _get_cmdline_unix_ps(): Retrieves command line using ps command
12+
- _get_cmdline_unix(): Dispatches to appropriate Unix method
13+
- _get_cmdline(): Main entry point with platform detection and error handling
14+
15+
"""
16+
17+
import os
18+
import subprocess
19+
import sys
20+
from typing import Generator
21+
from unittest.mock import Mock, mock_open, patch
22+
23+
import pytest
24+
25+
from instana.fsm import TheMachine
26+
27+
28+
class TestTheMachineCmdline:
29+
"""Test suite for TheMachine cmdline-related methods."""
30+
31+
@pytest.fixture(autouse=True)
32+
def _resource(self) -> Generator[None, None, None]:
33+
"""Setup and teardown for each test."""
34+
with patch("instana.fsm.TheMachine.__init__", return_value=None):
35+
self.machine = TheMachine(Mock())
36+
yield
37+
38+
@pytest.mark.parametrize(
39+
"cmdline_input,expected_output",
40+
[
41+
(
42+
"C:\\Python\\python.exe script.py arg1 arg2",
43+
["C:\\Python\\python.exe", "script.py", "arg1", "arg2"],
44+
),
45+
(
46+
"python.exe -m module --flag value",
47+
["python.exe", "-m", "module", "--flag", "value"],
48+
),
49+
("single_command", ["single_command"]),
50+
(
51+
"cmd.exe /c echo hello",
52+
["cmd.exe", "/c", "echo", "hello"],
53+
),
54+
],
55+
ids=[
56+
"full_path_with_args",
57+
"python_module_with_flags",
58+
"single_command",
59+
"cmd_with_subcommand",
60+
],
61+
)
62+
def test_get_cmdline_windows(
63+
self, cmdline_input: str, expected_output: list, mocker
64+
) -> None:
65+
"""Test _get_cmdline_windows with various command line formats."""
66+
mocker.patch(
67+
"ctypes.windll",
68+
create=True,
69+
)
70+
71+
with patch("ctypes.windll.kernel32.GetCommandLineW") as mock_get_cmdline:
72+
mock_get_cmdline.return_value = cmdline_input
73+
result = self.machine._get_cmdline_windows()
74+
assert result == expected_output
75+
76+
def test_get_cmdline_windows_empty_string(self, mocker) -> None:
77+
"""Test _get_cmdline_windows with empty command line."""
78+
mocker.patch(
79+
"ctypes.windll",
80+
create=True,
81+
)
82+
83+
with patch("ctypes.windll.kernel32.GetCommandLineW") as mock_get_cmdline:
84+
mock_get_cmdline.return_value = ""
85+
result = self.machine._get_cmdline_windows()
86+
assert result == []
87+
88+
@pytest.mark.parametrize(
89+
"proc_content,expected_output",
90+
[
91+
(
92+
"python\x00script.py\x00arg1\x00arg2\x00",
93+
["python", "script.py", "arg1", "arg2", ""],
94+
),
95+
(
96+
"/usr/bin/python3\x00-m\x00flask\x00run\x00",
97+
["/usr/bin/python3", "-m", "flask", "run", ""],
98+
),
99+
("gunicorn\x00app:app\x00", ["gunicorn", "app:app", ""]),
100+
("/usr/bin/python\x00", ["/usr/bin/python", ""]),
101+
(
102+
"python3\x00-c\x00print('hello')\x00",
103+
["python3", "-c", "print('hello')", ""],
104+
),
105+
],
106+
ids=[
107+
"basic_script_with_args",
108+
"python_module",
109+
"gunicorn_app",
110+
"single_executable",
111+
"python_command",
112+
],
113+
)
114+
def test_get_cmdline_linux_proc(
115+
self, proc_content: str, expected_output: list
116+
) -> None:
117+
"""Test _get_cmdline_linux_proc with various /proc/self/cmdline formats."""
118+
with patch("builtins.open", mock_open(read_data=proc_content)):
119+
result = self.machine._get_cmdline_linux_proc()
120+
assert result == expected_output
121+
122+
def test_get_cmdline_linux_proc_file_not_found(self) -> None:
123+
"""Test _get_cmdline_linux_proc when file doesn't exist."""
124+
with patch("builtins.open", side_effect=FileNotFoundError()):
125+
with pytest.raises(FileNotFoundError):
126+
self.machine._get_cmdline_linux_proc()
127+
128+
def test_get_cmdline_linux_proc_permission_error(self) -> None:
129+
"""Test _get_cmdline_linux_proc with permission error."""
130+
with patch("builtins.open", side_effect=PermissionError()):
131+
with pytest.raises(PermissionError):
132+
self.machine._get_cmdline_linux_proc()
133+
134+
@pytest.mark.parametrize(
135+
"ps_output,expected_output",
136+
[
137+
(
138+
b"COMMAND\npython script.py arg1 arg2\n",
139+
["python script.py arg1 arg2"],
140+
),
141+
(
142+
b"COMMAND\n/usr/bin/python3 -m flask run\n",
143+
["/usr/bin/python3 -m flask run"],
144+
),
145+
(b"COMMAND\ngunicorn app:app\n", ["gunicorn app:app"]),
146+
(b"COMMAND\npython\n", ["python"]),
147+
],
148+
ids=[
149+
"script_with_args",
150+
"python_module",
151+
"gunicorn",
152+
"single_command",
153+
],
154+
)
155+
def test_get_cmdline_unix_ps(self, ps_output: bytes, expected_output: list) -> None:
156+
"""Test _get_cmdline_unix_ps with various ps command outputs."""
157+
mock_proc = Mock()
158+
mock_proc.communicate.return_value = (ps_output, b"")
159+
160+
with patch("subprocess.Popen", return_value=mock_proc) as mock_popen:
161+
result = self.machine._get_cmdline_unix_ps(1234)
162+
assert result == expected_output
163+
mock_popen.assert_called_once_with(
164+
["ps", "-p", "1234", "-o", "args"], stdout=subprocess.PIPE
165+
)
166+
167+
def test_get_cmdline_unix_ps_with_different_pid(self) -> None:
168+
"""Test _get_cmdline_unix_ps with different PID values."""
169+
mock_proc = Mock()
170+
mock_proc.communicate.return_value = (b"COMMAND\ntest_process\n", b"")
171+
172+
with patch("subprocess.Popen", return_value=mock_proc) as mock_popen:
173+
result = self.machine._get_cmdline_unix_ps(9999)
174+
assert result == ["test_process"]
175+
mock_popen.assert_called_once_with(
176+
["ps", "-p", "9999", "-o", "args"], stdout=subprocess.PIPE
177+
)
178+
179+
def test_get_cmdline_unix_ps_empty_output(self) -> None:
180+
"""Test _get_cmdline_unix_ps with empty ps output."""
181+
mock_proc = Mock()
182+
mock_proc.communicate.return_value = (b"COMMAND\n\n", b"")
183+
184+
with patch("subprocess.Popen", return_value=mock_proc):
185+
result = self.machine._get_cmdline_unix_ps(1234)
186+
assert result == [""]
187+
188+
def test_get_cmdline_unix_ps_subprocess_error(self) -> None:
189+
"""Test _get_cmdline_unix_ps when subprocess fails."""
190+
with patch(
191+
"subprocess.Popen", side_effect=subprocess.SubprocessError("Test error")
192+
):
193+
with pytest.raises(subprocess.SubprocessError):
194+
self.machine._get_cmdline_unix_ps(1234)
195+
196+
@pytest.mark.parametrize(
197+
"proc_exists,proc_content,expected_output",
198+
[
199+
(
200+
True,
201+
"python\x00script.py\x00",
202+
["python", "script.py", ""],
203+
),
204+
(
205+
False,
206+
None,
207+
["ps_output"],
208+
),
209+
],
210+
ids=["proc_exists", "proc_not_exists"],
211+
)
212+
def test_get_cmdline_unix(
213+
self, proc_exists: bool, proc_content: str, expected_output: list
214+
) -> None:
215+
"""Test _get_cmdline_unix with and without /proc filesystem."""
216+
with patch("os.path.isfile", return_value=proc_exists):
217+
if proc_exists:
218+
with patch("builtins.open", mock_open(read_data=proc_content)):
219+
result = self.machine._get_cmdline_unix(1234)
220+
assert result == expected_output
221+
else:
222+
mock_proc = Mock()
223+
mock_proc.communicate.return_value = (b"COMMAND\nps_output\n", b"")
224+
with patch("subprocess.Popen", return_value=mock_proc):
225+
result = self.machine._get_cmdline_unix(1234)
226+
assert result == expected_output
227+
228+
def test_get_cmdline_unix_proc_file_check(self) -> None:
229+
"""Test _get_cmdline_unix checks for /proc/self/cmdline correctly."""
230+
with patch("os.path.isfile") as mock_isfile:
231+
mock_isfile.return_value = True
232+
with patch("builtins.open", mock_open(read_data="test\x00")):
233+
self.machine._get_cmdline_unix(1234)
234+
mock_isfile.assert_called_once_with("/proc/self/cmdline")
235+
236+
@pytest.mark.parametrize(
237+
"is_windows_value,expected_method",
238+
[
239+
(True, "_get_cmdline_windows"),
240+
(False, "_get_cmdline_unix"),
241+
],
242+
ids=["windows", "unix"],
243+
)
244+
def test_get_cmdline_platform_detection(
245+
self, is_windows_value: bool, expected_method: str
246+
) -> None:
247+
"""Test _get_cmdline correctly detects platform and calls appropriate method."""
248+
with patch("instana.fsm.is_windows", return_value=is_windows_value):
249+
if is_windows_value:
250+
with patch.object(
251+
self.machine, "_get_cmdline_windows", return_value=["windows_cmd"]
252+
) as mock_method:
253+
result = self.machine._get_cmdline(1234)
254+
assert result == ["windows_cmd"]
255+
mock_method.assert_called_once()
256+
else:
257+
with patch.object(
258+
self.machine, "_get_cmdline_unix", return_value=["unix_cmd"]
259+
) as mock_method:
260+
result = self.machine._get_cmdline(1234)
261+
assert result == ["unix_cmd"]
262+
mock_method.assert_called_once_with(1234)
263+
264+
def test_get_cmdline_windows_exception_fallback(self) -> None:
265+
"""Test _get_cmdline falls back to sys.argv on Windows exception."""
266+
with patch("instana.fsm.is_windows", return_value=True), patch.object(
267+
self.machine, "_get_cmdline_windows", side_effect=Exception("Test error")
268+
), patch("instana.fsm.logger.debug") as mock_logger:
269+
result = self.machine._get_cmdline(1234)
270+
assert result == sys.argv
271+
mock_logger.assert_called_once()
272+
273+
def test_get_cmdline_unix_exception_fallback(self) -> None:
274+
"""Test _get_cmdline falls back to sys.argv on Unix exception."""
275+
with patch("instana.fsm.is_windows", return_value=False), patch.object(
276+
self.machine, "_get_cmdline_unix", side_effect=Exception("Test error")
277+
), patch("instana.fsm.logger.debug") as mock_logger:
278+
result = self.machine._get_cmdline(1234)
279+
assert result == sys.argv
280+
mock_logger.assert_called_once()
281+
282+
@pytest.mark.parametrize(
283+
"exception_type",
284+
[
285+
OSError,
286+
IOError,
287+
PermissionError,
288+
FileNotFoundError,
289+
RuntimeError,
290+
],
291+
ids=[
292+
"OSError",
293+
"IOError",
294+
"PermissionError",
295+
"FileNotFoundError",
296+
"RuntimeError",
297+
],
298+
)
299+
def test_get_cmdline_various_exceptions(self, exception_type: type) -> None:
300+
"""Test _get_cmdline handles various exception types gracefully."""
301+
with patch("instana.fsm.is_windows", return_value=False), patch.object(
302+
self.machine, "_get_cmdline_unix", side_effect=exception_type("Test error")
303+
):
304+
result = self.machine._get_cmdline(1234)
305+
assert result == sys.argv
306+
307+
def test_get_cmdline_with_actual_pid(self) -> None:
308+
"""Test _get_cmdline with actual process ID."""
309+
current_pid = os.getpid()
310+
with patch("instana.fsm.is_windows", return_value=False), patch.object(
311+
self.machine, "_get_cmdline_unix", return_value=["test_cmd"]
312+
) as mock_method:
313+
result = self.machine._get_cmdline(current_pid)
314+
assert result == ["test_cmd"]
315+
mock_method.assert_called_once_with(current_pid)
316+
317+
def test_get_cmdline_windows_with_quotes(self, mocker) -> None:
318+
"""Test _get_cmdline_windows handles command lines with quotes."""
319+
cmdline_with_quotes = '"C:\\Program Files\\Python\\python.exe" "my script.py"'
320+
mocker.patch(
321+
"ctypes.windll",
322+
create=True,
323+
)
324+
325+
with patch("ctypes.windll.kernel32.GetCommandLineW") as mock_get_cmdline:
326+
mock_get_cmdline.return_value = cmdline_with_quotes
327+
result = self.machine._get_cmdline_windows()
328+
# Note: Simple split() doesn't handle quotes properly, this tests current behavior
329+
assert isinstance(result, list)
330+
assert len(result) > 0
331+
332+
def test_get_cmdline_linux_proc_with_empty_args(self) -> None:
333+
"""Test _get_cmdline_linux_proc with command that has empty arguments."""
334+
proc_content = "python\x00\x00\x00"
335+
with patch("builtins.open", mock_open(read_data=proc_content)):
336+
result = self.machine._get_cmdline_linux_proc()
337+
assert result == ["python", "", "", ""]
338+
339+
def test_get_cmdline_unix_ps_with_multiline_output(self) -> None:
340+
"""Test _get_cmdline_unix_ps handles multiline ps output correctly."""
341+
ps_output = b"COMMAND\npython script.py\nextra line\n"
342+
mock_proc = Mock()
343+
mock_proc.communicate.return_value = (ps_output, b"")
344+
345+
with patch("subprocess.Popen", return_value=mock_proc):
346+
result = self.machine._get_cmdline_unix_ps(1234)
347+
# Should only take the second line (index 1)
348+
assert result == ["python script.py"]
349+
350+
def test_get_cmdline_unix_ps_with_special_characters(self) -> None:
351+
"""Test _get_cmdline_unix_ps with special characters in command."""
352+
ps_output = b"COMMAND\npython -c 'print(\"hello\")'\n"
353+
mock_proc = Mock()
354+
mock_proc.communicate.return_value = (ps_output, b"")
355+
356+
with patch("subprocess.Popen", return_value=mock_proc):
357+
result = self.machine._get_cmdline_unix_ps(1234)
358+
assert result == ["python -c 'print(\"hello\")'"]
359+
360+
def test_get_cmdline_linux_proc_with_unicode(self) -> None:
361+
"""Test _get_cmdline_linux_proc with unicode characters."""
362+
proc_content = "python\x00script_café.py\x00"
363+
with patch("builtins.open", mock_open(read_data=proc_content)):
364+
result = self.machine._get_cmdline_linux_proc()
365+
assert "script_café.py" in result
366+
367+
@pytest.mark.parametrize(
368+
"pid_value",
369+
[1, 100, 9999, 65535],
370+
ids=["pid_1", "pid_100", "pid_9999", "pid_max"],
371+
)
372+
def test_get_cmdline_unix_ps_with_various_pids(self, pid_value: int) -> None:
373+
"""Test _get_cmdline_unix_ps with various PID values."""
374+
mock_proc = Mock()
375+
mock_proc.communicate.return_value = (b"COMMAND\ntest\n", b"")
376+
377+
with patch("subprocess.Popen", return_value=mock_proc) as mock_popen:
378+
self.machine._get_cmdline_unix_ps(pid_value)
379+
mock_popen.assert_called_once_with(
380+
["ps", "-p", str(pid_value), "-o", "args"], stdout=subprocess.PIPE
381+
)
382+
383+
384+
# Made with Bob

0 commit comments

Comments
 (0)