Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 5 additions & 2 deletions .pylintrc
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,9 @@ expected-line-ending-format = LF
# Allow unused arguments in test fixtures (pytest mocks/fixtures commonly have unused params)
dummy-variables-rgx = ^_|^mock_|^fixture_|^suppress_|^monkeypatch|^temp_|^fake_

# Disable unused-argument check for test files where pytest fixtures intentionally have unused parameters
# Disable specific rules for test files as they need to operate specifically and differently from production code
[MESSAGES CONTROL:tests/python/test_*.py]
disable = W0613
disable =
W0212, # Access to a protected member
W0613, # Unused argument
W0621 # Redefining name from outer scopes
6 changes: 5 additions & 1 deletion shared/python/apimrequests.py
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,11 @@ def _multiRequest(self, method: HTTP_VERB, path: str, runs: int, headers: list[a

session = requests.Session()

session.headers.update(self.headers.copy())
merged_headers = self.headers.copy()
if headers:
merged_headers.update(headers)

session.headers.update(merged_headers)

try:
if msg:
Expand Down
97 changes: 96 additions & 1 deletion tests/python/test_apimrequests.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
"""Tests for apimrequests helpers and request behavior."""

from unittest.mock import patch, MagicMock
import requests
import pytest
Expand Down Expand Up @@ -154,13 +156,15 @@ def test_request_header_merging():
@pytest.mark.http
def test_init_missing_url():
# Negative: missing URL should raise TypeError
init = ApimRequests.__init__
with pytest.raises(TypeError):
ApimRequests() # pylint: disable=no-value-for-parameter
init(ApimRequests)

@pytest.mark.http
def test_print_response_code_edge():
apim = make_apim()
class DummyResponse:
"""Response stub for non-2xx status formatting."""
status_code = 302
reason = 'Found'
with patch('apimrequests.print_val') as mock_print_val:
Expand Down Expand Up @@ -510,12 +514,14 @@ def test_print_response_200_valid_json(apim, apimrequests_patches):
def test_print_response_code_success_and_error(apim, apimrequests_patches):
"""Test _print_response_code color formatting for success and error codes."""
class DummyResponse:
"""Response stub for successful status formatting."""
status_code = 200
reason = 'OK'

apim._print_response_code(DummyResponse())

class ErrorResponse:
"""Response stub for error status formatting."""
status_code = 500
reason = 'Server Error'

Expand Down Expand Up @@ -725,6 +731,7 @@ def test_single_post_async_non_json_response(apim, apimrequests_patches):
def test_print_response_code_2xx_non_200(apim, apimrequests_patches):
"""Test _print_response_code with 2xx status codes other than 200."""
class DummyResponse:
"""Response stub for 2xx non-200 status formatting."""
status_code = 201
reason = 'Created'

Expand All @@ -741,6 +748,7 @@ class DummyResponse:
def test_print_response_code_3xx(apim, apimrequests_patches):
"""Test _print_response_code with 3xx redirect status codes."""
class DummyResponse:
"""Response stub for 3xx status formatting."""
status_code = 301
reason = 'Moved Permanently'

Expand Down Expand Up @@ -987,3 +995,90 @@ def test_multi_request_session_exception_on_request(apim):

# Verify session was closed even after exception
mock_session.close.assert_called_once()


@pytest.mark.unit
def test_multi_request_merges_custom_headers(apim):
"""Test _multiRequest merges passed headers with default headers."""
custom_headers = {'X-Custom-Header': 'custom-value', 'X-Request-Id': '123'}

with patch('apimrequests.requests.Session') as mock_session_cls:
mock_session = MagicMock()
response = create_mock_http_response(json_data={'result': 'ok'})
mock_session.request.return_value = response
mock_session_cls.return_value = mock_session

with patch.object(apim, '_print_response_code'):
apim._multiRequest(HTTP_VERB.GET, '/test', 1, headers=custom_headers, printResponse=False)

# Verify headers.update was called with merged headers
update_call_args = mock_session.headers.update.call_args
merged_headers = update_call_args[0][0]

# Check custom headers are included
assert merged_headers['X-Custom-Header'] == 'custom-value'
assert merged_headers['X-Request-Id'] == '123'
# Check default headers are still there
assert 'Accept' in merged_headers
assert merged_headers['Accept'] == 'application/json'
assert SUBSCRIPTION_KEY_PARAMETER_NAME in merged_headers


@pytest.mark.unit
def test_multi_get_merges_custom_headers(apim):
"""Test multiGet merges custom headers into requests."""
custom_headers = {'X-Custom-Header': 'custom-value'}

with patch('apimrequests.requests.Session') as mock_session_cls:
mock_session = MagicMock()
response = create_mock_http_response(json_data={'result': 'ok'})
mock_session.request.return_value = response
mock_session_cls.return_value = mock_session

with patch.object(apim, '_print_response_code'):
result = apim.multiGet('/test', runs=2, headers=custom_headers, printResponse=False)
def test_single_request_merges_custom_headers(apim):
"""Test singleGet merges custom headers with default headers."""
custom_headers = {'X-Custom-Header': 'test-value'}

mock_response = create_mock_http_response(json_data={'result': 'ok'})

with patch('apimrequests.requests.request') as mock_request:
mock_request.return_value = mock_response

with patch.object(apim, '_print_response'):
apim.singleGet('/test', headers=custom_headers, printResponse=True)

# Verify merged headers were passed to request
call_kwargs = mock_request.call_args[1]
merged_headers = call_kwargs['headers']

assert merged_headers['X-Custom-Header'] == 'test-value'
assert 'Accept' in merged_headers
assert SUBSCRIPTION_KEY_PARAMETER_NAME in merged_headers


@pytest.mark.unit
def test_multi_request_custom_headers_do_not_affect_other_runs(apim):
"""Test that custom headers persist across multiple runs in multiGet."""
custom_headers = {'X-Request-Id': 'same-id'}

with patch('apimrequests.requests.Session') as mock_session_cls:
mock_session = MagicMock()
response = create_mock_http_response(json_data={'result': 'ok'})
mock_session.request.return_value = response
mock_session_cls.return_value = mock_session

with patch.object(apim, '_print_response_code'):
result = apim.multiGet('/test', runs=3, headers=custom_headers, printResponse=False)

# Verify headers.update was called once with merged headers
assert mock_session.headers.update.call_count == 1

# Verify headers contain custom header
update_call_args = mock_session.headers.update.call_args
merged_headers = update_call_args[0][0]
assert merged_headers['X-Request-Id'] == 'same-id'

# Verify all 3 runs completed
assert len(result) == 3
28 changes: 19 additions & 9 deletions tests/python/test_local_setup.py
Original file line number Diff line number Diff line change
Expand Up @@ -209,7 +209,7 @@ def test_get_project_root_finds_indicators(tmp_path: Path):
assert all((tmp_path / indicator).exists() for indicator in ["README.md", "bicepconfig.json"])


def test_setup_python_path(temp_project_root: Path, monkeypatch: pytest.MonkeyPatch):
def test_setup_python_path(temp_project_root: Path):
"""Test setup_python_path adds shared path to sys.path."""
original_sys_path = sys.path.copy()

Expand All @@ -223,7 +223,7 @@ def test_setup_python_path(temp_project_root: Path, monkeypatch: pytest.MonkeyPa
sys.path[:] = original_sys_path


def test_setup_python_path_already_in_path(temp_project_root: Path, monkeypatch: pytest.MonkeyPatch):
def test_setup_python_path_already_in_path(temp_project_root: Path):
"""Test setup_python_path doesn't duplicate existing paths."""
original_sys_path = sys.path.copy()

Expand Down Expand Up @@ -602,15 +602,15 @@ def test_generate_env_file_creates_missing_env(temp_project_root: Path) -> None:
# Tests for Jupyter kernel setup
# ============================================================

def test_install_jupyter_kernel_success(monkeypatch: pytest.MonkeyPatch):
def test_install_jupyter_kernel_success():
"""Test install_jupyter_kernel succeeds with ipykernel available."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = Mock(returncode=0)
result = sps.install_jupyter_kernel()
assert result is True


def test_install_jupyter_kernel_ipykernel_not_installed(monkeypatch: pytest.MonkeyPatch):
def test_install_jupyter_kernel_ipykernel_not_installed():
"""Test install_jupyter_kernel installs ipykernel if missing."""
call_count = [0]

Expand Down Expand Up @@ -666,7 +666,7 @@ def mock_run(args, **kwargs):
assert [sys.executable, "-m", "pip", "install", "ipykernel"] in call_log


def test_install_jupyter_kernel_registration_fails(monkeypatch: pytest.MonkeyPatch):
def test_install_jupyter_kernel_registration_fails():
"""Test install_jupyter_kernel handles registration failures."""
call_count = [0]

Expand All @@ -683,7 +683,7 @@ def mock_run(*args, **kwargs):
assert result is False


def test_validate_kernel_setup_kernel_found(monkeypatch: pytest.MonkeyPatch):
def test_validate_kernel_setup_kernel_found():
"""Test validate_kernel_setup returns True when kernel is found."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = Mock(
Expand All @@ -694,7 +694,7 @@ def test_validate_kernel_setup_kernel_found(monkeypatch: pytest.MonkeyPatch):
assert result is True


def test_validate_kernel_setup_kernel_not_found(monkeypatch: pytest.MonkeyPatch):
def test_validate_kernel_setup_kernel_not_found():
"""Test validate_kernel_setup returns False when kernel not found."""
with patch("subprocess.run") as mock_run:
mock_run.return_value = Mock(
Expand All @@ -705,15 +705,15 @@ def test_validate_kernel_setup_kernel_not_found(monkeypatch: pytest.MonkeyPatch)
assert result is False


def test_validate_kernel_setup_jupyter_error(monkeypatch: pytest.MonkeyPatch):
def test_validate_kernel_setup_jupyter_error():
"""Test validate_kernel_setup handles Jupyter command errors."""
with patch("subprocess.run") as mock_run:
mock_run.side_effect = subprocess.CalledProcessError(1, "jupyter")
result = sps.validate_kernel_setup()
assert result is False


def test_validate_kernel_setup_jupyter_not_found(monkeypatch: pytest.MonkeyPatch):
def test_validate_kernel_setup_jupyter_not_found():
"""Test validate_kernel_setup handles missing Jupyter."""
with patch("subprocess.run") as mock_run:
mock_run.side_effect = FileNotFoundError()
Expand Down Expand Up @@ -1310,6 +1310,7 @@ def test_force_kernel_consistency_exception_on_write(temp_project_root: Path, mo

def test_setup_complete_environment_summary_messages(temp_project_root: Path, monkeypatch: pytest.MonkeyPatch, capsys):
"""Test setup_complete_environment displays summary with mixed results."""
monkeypatch.setattr(sps, "check_uv_installed", lambda: False)
monkeypatch.setattr(sps, "check_azure_cli_installed", lambda: True)
monkeypatch.setattr(sps, "check_bicep_cli_installed", lambda: True)
monkeypatch.setattr(sps, "check_azure_providers_registered", lambda: True)
Expand Down Expand Up @@ -1672,9 +1673,14 @@ def test_install_jupyter_kernel_ipykernel_already_installed(monkeypatch: pytest.

def test_setup_complete_environment_with_missing_bicep(temp_project_root: Path, monkeypatch: pytest.MonkeyPatch):
"""Test setup_complete_environment stops when Bicep CLI is missing."""
monkeypatch.setattr(sps, "check_uv_installed", lambda: False)
monkeypatch.setattr(sps, "check_azure_cli_installed", lambda: True)
monkeypatch.setattr(sps, "check_bicep_cli_installed", lambda: False)
monkeypatch.setattr(sps, "check_azure_providers_registered", lambda: True)
monkeypatch.setattr(sps, "generate_env_file", lambda: None)
monkeypatch.setattr(sps, "install_jupyter_kernel", lambda: True)
monkeypatch.setattr(sps, "create_vscode_settings", lambda: True)
monkeypatch.setattr(sps, "force_kernel_consistency", lambda: True)

# Should return early without full setup
sps.setup_complete_environment()
Expand Down Expand Up @@ -2589,6 +2595,7 @@ def test_generate_env_file_with_malformed_lines(temp_project_root: Path):

def test_setup_complete_environment_all_pass(monkeypatch: pytest.MonkeyPatch):
"""Test setup_complete_environment when all checks pass."""
monkeypatch.setattr(sps, "check_uv_installed", lambda: False)
with patch.object(sps, "check_azure_cli_installed", return_value=True):
with patch.object(sps, "check_bicep_cli_installed", return_value=True):
with patch.object(sps, "check_azure_providers_registered", return_value=True):
Expand All @@ -2615,6 +2622,7 @@ def test_setup_complete_environment_azure_cli_fails(monkeypatch: pytest.MonkeyPa

def test_setup_complete_environment_kernel_fails(monkeypatch: pytest.MonkeyPatch):
"""Test setup_complete_environment when kernel registration fails."""
monkeypatch.setattr(sps, "check_uv_installed", lambda: False)
with patch.object(sps, "check_azure_cli_installed", return_value=True):
with patch.object(sps, "check_bicep_cli_installed", return_value=True):
with patch.object(sps, "check_azure_providers_registered", return_value=True):
Expand Down Expand Up @@ -3074,6 +3082,7 @@ def test_ensure_utf8_streams_without_reconfigure(monkeypatch: pytest.MonkeyPatch
"""Test _ensure_utf8_streams when streams lack reconfigure attribute."""

class DummyStream:
"""Mock stream without reconfigure method."""
encoding = None

original_stdout, original_stderr = sys.stdout, sys.stderr
Expand All @@ -3092,6 +3101,7 @@ def test_ensure_utf8_streams_reconfigure_failure(monkeypatch: pytest.MonkeyPatch
"""Test _ensure_utf8_streams when reconfigure raises an exception."""

class BrokenStream:
"""Mock stream with reconfigure that raises an exception."""
encoding = None

def reconfigure(self, **kwargs):
Expand Down
Loading
Loading