|
| 1 | +# (c) Copyright IBM Corp. 2025 |
| 2 | +# Assisted by watsonx Code Assistant |
| 3 | + |
| 4 | +import logging |
| 5 | +import os |
| 6 | +import sys |
| 7 | +from typing import TYPE_CHECKING, Generator, List, Union |
| 8 | + |
| 9 | +import pytest |
| 10 | + |
| 11 | +from instana.util.runtime import ( |
| 12 | + determine_service_name, |
| 13 | + get_proc_cmdline, |
| 14 | + get_py_source, |
| 15 | + get_runtime_env_info, |
| 16 | + log_runtime_env_info, |
| 17 | +) |
| 18 | + |
| 19 | +if TYPE_CHECKING: |
| 20 | + from pytest import LogCaptureFixture |
| 21 | + from pytest_mock import MockerFixture |
| 22 | + |
| 23 | + |
| 24 | +def test_get_py_source(tmp_path) -> None: |
| 25 | + """Test the get_py_source.""" |
| 26 | + filename = "temp_file.py" |
| 27 | + file_contents = "print('Hello, World!')\n" |
| 28 | + expected_output = {"data": file_contents} |
| 29 | + |
| 30 | + # Create a temporary file for testing purposes. |
| 31 | + temp_file = tmp_path / filename |
| 32 | + temp_file.write_text(file_contents) |
| 33 | + |
| 34 | + result = get_py_source(f"{tmp_path}/{filename}") |
| 35 | + assert result == expected_output, f"Expected {expected_output}, but got {result}" |
| 36 | + |
| 37 | + |
| 38 | +@pytest.mark.parametrize( |
| 39 | + "filename, expected_output", |
| 40 | + [ |
| 41 | + ( |
| 42 | + "non_existent_file.py", |
| 43 | + {"error": "[Errno 2] No such file or directory: 'non_existent_file.py'"} |
| 44 | + ), |
| 45 | + ("temp_file.txt", {"error": "Only Python source files are allowed. (*.py)"}), |
| 46 | + ], |
| 47 | +) |
| 48 | +def test_get_py_source_error(filename, expected_output) -> None: |
| 49 | + """Test the get_py_source function with various scenarios with errors.""" |
| 50 | + result = get_py_source(filename) |
| 51 | + assert result == expected_output, f"Expected {expected_output}, but got {result}" |
| 52 | + |
| 53 | + |
| 54 | +def test_get_py_source_exception(mocker) -> None: |
| 55 | + """Test the get_py_source function with an exception scenario.""" |
| 56 | + exception_message = "No such file or directory" |
| 57 | + mocker.patch( |
| 58 | + "instana.util.runtime.get_py_source", side_effect=Exception(exception_message) |
| 59 | + ) |
| 60 | + |
| 61 | + with pytest.raises(Exception) as exc_info: |
| 62 | + get_py_source("/path/to/non_readable_file.py") |
| 63 | + assert str(exc_info.value) == exception_message, ( |
| 64 | + f"Expected {exception_message}, but got {exc_info.value}" |
| 65 | + ) |
| 66 | + |
| 67 | + |
| 68 | +@pytest.fixture() |
| 69 | +def _resource_determine_service_name_via_env_var() -> Generator[None, None, None]: |
| 70 | + """SetUp and TearDown""" |
| 71 | + # setup |
| 72 | + yield |
| 73 | + # teardown |
| 74 | + os.environ.pop("INSTANA_SERVICE_NAME", None) |
| 75 | + os.environ.pop("FLASK_APP", None) |
| 76 | + os.environ.pop("DJANGO_SETTINGS_MODULE", None) |
| 77 | + |
| 78 | + |
| 79 | +@pytest.mark.parametrize( |
| 80 | + "env_var, value, expected_output", |
| 81 | + [ |
| 82 | + ("INSTANA_SERVICE_NAME", "test_service", "test_service"), |
| 83 | + ("FLASK_APP", "test_flask_app.py", "test_flask_app.py"), |
| 84 | + ("DJANGO_SETTINGS_MODULE", "test_django_app.settings", "test_django_app"), |
| 85 | + ], |
| 86 | +) |
| 87 | +def test_determine_service_name_via_env_var( |
| 88 | + env_var: str, |
| 89 | + value: str, |
| 90 | + expected_output: str, |
| 91 | + _resource_determine_service_name_via_env_var: None, |
| 92 | +) -> None: |
| 93 | + # Test with multiple environment variables |
| 94 | + os.environ[env_var] = value |
| 95 | + sys.argv = ["something", "nothing"] |
| 96 | + assert determine_service_name() == expected_output |
| 97 | + |
| 98 | + |
| 99 | +@pytest.mark.parametrize( |
| 100 | + "web_browser, argv, expected_output", |
| 101 | + [ |
| 102 | + ("gunicorn", ["gunicorn", "djface.wsgi:app"], "gunicorn"), |
| 103 | + ( |
| 104 | + "uwsgi", |
| 105 | + [ |
| 106 | + "uwsgi", |
| 107 | + "--master", |
| 108 | + "--processes", |
| 109 | + "4", |
| 110 | + "--threads", |
| 111 | + "2", |
| 112 | + "djface.wsgi:app", |
| 113 | + ], |
| 114 | + "uWSGI master", |
| 115 | + ), |
| 116 | + ], |
| 117 | +) |
| 118 | +def test_determine_service_name_via_web_browser( |
| 119 | + web_browser: str, |
| 120 | + argv: List[str], |
| 121 | + expected_output: str, |
| 122 | + _resource_determine_service_name_via_env_var: None, |
| 123 | + mocker: "MockerFixture", |
| 124 | +) -> None: |
| 125 | + mocker.patch("instana.util.runtime.get_proc_cmdline", return_value="python") |
| 126 | + mocker.patch("os.getpid", return_value=12345) |
| 127 | + sys.argv = argv |
| 128 | + assert determine_service_name() == expected_output |
| 129 | + |
| 130 | + |
| 131 | +@pytest.mark.parametrize( |
| 132 | + "argv", |
| 133 | + [ |
| 134 | + (["python", "test_app.py", "arg1", "arg2"]), |
| 135 | + ([]), |
| 136 | + ], |
| 137 | +) |
| 138 | +def test_determine_service_name_via_cli_args( |
| 139 | + argv: List[str], |
| 140 | + _resource_determine_service_name_via_env_var: None, |
| 141 | + mocker: "MockerFixture", |
| 142 | +) -> None: |
| 143 | + mocker.patch("instana.util.runtime.get_proc_cmdline", return_value="python") |
| 144 | + sys.argv = argv |
| 145 | + # We check "python" in the return of determine_service_name() because this |
| 146 | + # can be the value "python3" |
| 147 | + assert "python" in determine_service_name() |
| 148 | + |
| 149 | + |
| 150 | +@pytest.mark.parametrize( |
| 151 | + "isatty, expected_output", |
| 152 | + [ |
| 153 | + (True, "Interactive Console"), |
| 154 | + (False, ""), |
| 155 | + ], |
| 156 | +) |
| 157 | +def test_determine_service_name_via_tty( |
| 158 | + isatty: bool, |
| 159 | + expected_output: str, |
| 160 | + _resource_determine_service_name_via_env_var: None, |
| 161 | + mocker: "MockerFixture", |
| 162 | +) -> None: |
| 163 | + sys.argv = [] |
| 164 | + sys.executable = "" |
| 165 | + sys.stdout.isatty = lambda: isatty |
| 166 | + assert determine_service_name() == expected_output |
| 167 | + |
| 168 | + |
| 169 | +@pytest.mark.parametrize( |
| 170 | + "as_string, expected", |
| 171 | + [ |
| 172 | + (False, ["python", "script.py", "arg1", "arg2"]), |
| 173 | + (True, "python script.py arg1 arg2"), |
| 174 | + ], |
| 175 | +) |
| 176 | +def test_get_proc_cmdline(as_string: bool, expected: Union[List[str], str], mocker: "MockerFixture") -> None: |
| 177 | + # Mock the proc filesystem presence |
| 178 | + mocker.patch("os.path.isfile", return_value="/proc/self/cmdline") |
| 179 | + # Mock the content of /proc/self/cmdline |
| 180 | + mocked_data = mocker.mock_open(read_data="python\0script.py\0arg1\0arg2\0") |
| 181 | + mocker.patch("builtins.open", mocked_data) |
| 182 | + |
| 183 | + assert get_proc_cmdline(as_string) == expected, f"Expected {expected}, but got {get_proc_cmdline(as_string)}" |
| 184 | + |
| 185 | + |
| 186 | +@pytest.mark.parametrize( |
| 187 | + "as_string, expected", |
| 188 | + [ |
| 189 | + (False, ["python"]), |
| 190 | + (True, "python"), |
| 191 | + ], |
| 192 | +) |
| 193 | +def test_get_proc_cmdline_no_proc_fs( |
| 194 | + as_string: bool, expected: Union[List[str], str], mocker: "MockerFixture" |
| 195 | +): |
| 196 | + # Mock the proc filesystem absence |
| 197 | + mocker.patch("os.path.isfile", return_value=False) |
| 198 | + assert get_proc_cmdline(as_string) == expected |
| 199 | + |
| 200 | + |
| 201 | + |
| 202 | +def test_get_runtime_env_info(mocker: "MockerFixture") -> None: |
| 203 | + """Test the get_runtime_env_info function.""" |
| 204 | + expected_output = ("x86_64", "3.13.5") |
| 205 | + |
| 206 | + mocker.patch("platform.machine", return_value=expected_output[0]) |
| 207 | + mocker.patch("platform.python_version", return_value=expected_output[1]) |
| 208 | + |
| 209 | + machine, py_version = get_runtime_env_info() |
| 210 | + assert machine == expected_output[0] |
| 211 | + assert py_version == expected_output[1] |
| 212 | + |
| 213 | + |
| 214 | +def test_log_runtime_env_info(mocker: "MockerFixture", caplog: "LogCaptureFixture") -> None: |
| 215 | + """Test the log_runtime_env_info function.""" |
| 216 | + expected_output = ("x86_64", "3.13.5") |
| 217 | + caplog.set_level(logging.DEBUG, logger="instana") |
| 218 | + |
| 219 | + mocker.patch("platform.machine", return_value=expected_output[0]) |
| 220 | + mocker.patch("platform.python_version", return_value=expected_output[1]) |
| 221 | + |
| 222 | + log_runtime_env_info() |
| 223 | + assert ( |
| 224 | + f"Runtime environment: Machine: {expected_output[0]}, Python version: {expected_output[1]}" |
| 225 | + in caplog.messages |
| 226 | + ) |
0 commit comments