diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index f674fef..d602849 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -1,6 +1,7 @@ name: Python Tests on: + workflow_dispatch: push: branches: [ main ] pull_request: @@ -8,6 +9,8 @@ on: permissions: contents: read + checks: write + pull-requests: write jobs: test: @@ -17,14 +20,53 @@ jobs: python-version: [ '3.12', '3.13' ] steps: - uses: actions/checkout@v4 + - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v5 with: python-version: ${{ matrix.python-version }} + - name: Install dependencies run: | python -m pip install --upgrade pip pip install -r requirements.txt - - name: Run tests + + - name: Install pytest and pytest-cov + run: | + pip install pytest pytest-cov + + # Run tests with continue-on-error so that coverage and PR comments are always published. + # The final step will explicitly fail the job if any test failed, ensuring PRs cannot be merged with failing tests. + - name: Run pytest with coverage and generate JUnit XML + run: | + PYTHONPATH=$(pwd) COVERAGE_FILE=tests/python/.coverage-${{ matrix.python-version }} pytest --cov=shared/python --cov-config=tests/python/.coveragerc --cov-report=html:tests/python/htmlcov-${{ matrix.python-version }} --junitxml=tests/python/junit-${{ matrix.python-version }}.xml tests/python/ + continue-on-error: true + + - name: Upload coverage HTML report + uses: actions/upload-artifact@v4 + with: + name: coverage-html-${{ matrix.python-version }} + path: tests/python/htmlcov-${{ matrix.python-version }}/ + + - name: Upload JUnit test results + uses: actions/upload-artifact@v4 + with: + name: junit-results-${{ matrix.python-version }} + path: tests/python/junit-${{ matrix.python-version }}.xml + + - name: Publish Unit Test Results to PR + if: always() + uses: EnricoMi/publish-unit-test-result-action@v2 + with: + files: tests/python/junit-${{ matrix.python-version }}.xml + comment_title: Python ${{ matrix.python-version }} Test Results + + # Explicitly fail the job if any test failed (so PRs cannot be merged with failing tests). + # This runs after all reporting steps, meaning coverage and PR comments are always published. + - name: Fail if tests failed + if: always() run: | - python -m unittest discover shared/python-tests + if grep -q 'failures="[1-9]' tests/python/junit-${{ matrix.python-version }}.xml; then + echo "::error ::Unit tests failed. See above for details." + exit 1 + fi \ No newline at end of file diff --git a/.gitignore b/.gitignore index a8ecba6..3fe9874 100644 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,8 @@ labs-in-progress/ # Exclude sensitive or generated files .env + + +# Coverage data and reports +.coverage +tests/python/htmlcov/ diff --git a/README.md b/README.md index 126f7e8..23bbf39 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # Azure API Management Samples +[![Python Tests](https://github.com/Azure-Samples/Apim-Samples/actions/workflows/python-tests.yml/badge.svg?branch=main)](https://github.com/Azure-Samples/Apim-Samples/actions/workflows/python-tests.yml) + This repository provides a playground to safely experiment with and learn Azure API Management (APIM) policies in various architectures. ## Objectives @@ -87,11 +89,53 @@ The repo uses the bicep linter and has rules defined in `bicepconfig.json`. See We welcome contributions! Please consider forking the repo and creating issues and pull requests to share your samples. Please see [CONTRIBUTING.md](CONTRIBUTING.md) for details. Thank you! -### Testing -Python modules and files in the `shared` folder are supported with tests in the `python-tests` folder. They can be invoked manually via `pytest -v` but are also run upon push to GitHub origin. The `pytest.ini` file in the root sets up the test structure. +### Testing & Code Coverage + +Python modules in `shared/python` are covered by comprehensive unit tests located in `tests/python`. All tests use [pytest](https://docs.pytest.org/) and leverage modern pytest features, including custom markers for unit and HTTP tests. + +#### Running Tests Locally + +- **PowerShell (Windows):** + - Run all tests with coverage: `./tests/python/run_tests.ps1` +- **Shell (Linux/macOS):** + - Run all tests with coverage: `./tests/python/run_tests.sh` + +Both scripts: +- Run all tests in `tests/python` using pytest +- Generate a code coverage report (HTML output in `tests/python/htmlcov`) +- Store the raw coverage data in `tests/python/.coverage` + +You can also run tests manually and see details in the console: +```sh +pytest -v --cov=shared/python --cov-report=html:tests/python/htmlcov --cov-report=term tests/python +``` + +#### Viewing Coverage Reports + +After running tests, open `tests/python/htmlcov/index.html` in your browser to view detailed coverage information. + +#### Pytest Markers + +- `@pytest.mark.unit` โ€” marks a unit test +- `@pytest.mark.http` โ€” marks a test involving HTTP/mocking + +Markers are registered in `pytest.ini` to avoid warnings. + +#### Continuous Integration (CI) + +On every push or pull request, GitHub Actions will: +- Install dependencies +- Run all Python tests in `tests/python` with coverage +- Store the `.coverage` file in `tests/python` +- Upload the HTML coverage report as a workflow artifact for download + +#### Additional Notes + +- The `.gitignore` is configured to exclude coverage output and artifacts. +- All test and coverage features work both locally and in CI. -Additional information can be found [here](# https://docs.pytest.org/en/8.2.x/). +For more details on pytest usage, see the [pytest documentation](https://docs.pytest.org/en/8.2.x/). --- diff --git a/infrastructure/afd-apim/clean-up.ipynb b/infrastructure/afd-apim/clean-up.ipynb index 76b7959..e001442 100644 --- a/infrastructure/afd-apim/clean-up.ipynb +++ b/infrastructure/afd-apim/clean-up.ipynb @@ -19,9 +19,9 @@ "from apimtypes import INFRASTRUCTURE\n", "\n", "deployment = INFRASTRUCTURE.AFD_APIM_PE\n", - "index = 1\n", + "indexes = [1]\n", "\n", - "utils.cleanup_resources(deployment, utils.get_infra_rg_name(deployment, index))" + "utils.cleanup_infra_deployments(deployment, indexes)" ] } ], diff --git a/infrastructure/apim-aca/clean-up.ipynb b/infrastructure/apim-aca/clean-up.ipynb index a385e0f..937a3ea 100644 --- a/infrastructure/apim-aca/clean-up.ipynb +++ b/infrastructure/apim-aca/clean-up.ipynb @@ -19,9 +19,9 @@ "from apimtypes import INFRASTRUCTURE\n", "\n", "deployment = INFRASTRUCTURE.APIM_ACA\n", - "index = 1\n", + "indexes = [1]\n", "\n", - "utils.cleanup_resources(deployment, utils.get_infra_rg_name(deployment, index))" + "utils.cleanup_infra_deployments(deployment, indexes)" ] } ], diff --git a/infrastructure/simple-apim/clean-up.ipynb b/infrastructure/simple-apim/clean-up.ipynb index 8e106b4..fd26d85 100644 --- a/infrastructure/simple-apim/clean-up.ipynb +++ b/infrastructure/simple-apim/clean-up.ipynb @@ -19,9 +19,9 @@ "from apimtypes import INFRASTRUCTURE\n", "\n", "deployment = INFRASTRUCTURE.SIMPLE_APIM\n", - "index = 1\n", + "indexes = [1]\n", "\n", - "utils.cleanup_resources(deployment, utils.get_infra_rg_name(deployment, index))" + "utils.cleanup_infra_deployments(deployment, indexes)" ] } ], @@ -41,7 +41,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.12.9" + "version": "3.12.10" } }, "nbformat": 4, diff --git a/pytest.ini b/pytest.ini deleted file mode 100644 index 4756da7..0000000 --- a/pytest.ini +++ /dev/null @@ -1,7 +0,0 @@ -[pytest] -markers = - TestApimTypes: Tests related to the APIM types. -testpaths = - shared/python-tests -python_files = - test_*.py diff --git a/requirements.txt b/requirements.txt index a8a7f34..a2cf31f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,4 +2,5 @@ requests setuptools -pytest \ No newline at end of file +pytest +pytest-cov \ No newline at end of file diff --git a/shared/python-tests/__init__.py b/shared/python-tests/__init__.py deleted file mode 100644 index 2923141..0000000 --- a/shared/python-tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -# Makes shared/tests a package for unittest discovery diff --git a/shared/python-tests/test_apimtypes.py b/shared/python-tests/test_apimtypes.py deleted file mode 100644 index 6c0727d..0000000 --- a/shared/python-tests/test_apimtypes.py +++ /dev/null @@ -1,32 +0,0 @@ -# pylint: disable=C0114,C0115,C0116,W0212,W0621 - -# Execute with "pytest -v" in the root directory - -# https://docs.pytest.org/en/8.2.x/ - -""" -Unit tests for apimtypes.py -""" -import pytest, sys, os, unittest - -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..'))) - -########################################################################################################################################################## - -# Utility Functions - -# Test Fixtures - -########################################################################################################################################################## - -# Tests - -class TestApimTypes(unittest.TestCase): - @pytest.mark.apimtypes - - def test_sample(self) -> None: - """Sample test - replace with real tests.""" - self.assertTrue(True) - -if __name__ == "__main__": - unittest.main() diff --git a/shared/python/apimtypes.py b/shared/python/apimtypes.py index fa419af..6b65415 100644 --- a/shared/python/apimtypes.py +++ b/shared/python/apimtypes.py @@ -145,6 +145,13 @@ class APIOperation: # ------------------------------ def __init__(self, name: str, displayName: str, urlTemplate: str, method: HTTP_VERB, description: str, policyXml: str): + # Validate that method is a valid HTTP_VERB + if not isinstance(method, HTTP_VERB): + try: + method = HTTP_VERB(method) + except Exception: + raise ValueError(f"Invalid HTTP_VERB: {method}") + self.name = name self.displayName = displayName self.method = method diff --git a/shared/python/utils.py b/shared/python/utils.py index 7282aed..5cb2db6 100644 --- a/shared/python/utils.py +++ b/shared/python/utils.py @@ -262,14 +262,13 @@ def policy_xml_replacement(policy_xml_filepath: str) -> str: # Convert the XML to JSON format return policy_template_xml -# Cleans up resources associated with a deployment in a resource group -def cleanup_resources(deployment: str | INFRASTRUCTURE, rg_name: str ) -> None: +def _cleanup_resources(deployment_name: str, rg_name: str) -> None: """ Clean up resources associated with a deployment in a resource group. Deletes and purges Cognitive Services, API Management, and Key Vault resources, then deletes the resource group itself. Args: - deployment (str | INFRASTRUCTURE): The deployment name or enum value. + deployment_name (str): The deployment name (string). rg_name (str): The resource group name. Returns: @@ -278,8 +277,7 @@ def cleanup_resources(deployment: str | INFRASTRUCTURE, rg_name: str ) -> None: Raises: Exception: If an error occurs during cleanup. """ - - if not deployment: + if not deployment_name: print_error("Missing deployment name parameter.") return @@ -288,11 +286,6 @@ def cleanup_resources(deployment: str | INFRASTRUCTURE, rg_name: str ) -> None: return try: - if hasattr(deployment, 'value'): - deployment_name = deployment.value - else: - deployment_name = deployment - print_info(f"๐Ÿงน Cleaning up resource group '{rg_name}'...") # Show the deployment details @@ -336,34 +329,84 @@ def cleanup_resources(deployment: str | INFRASTRUCTURE, rg_name: str ) -> None: print(f"An error occurred during cleanup: {e}") traceback.print_exc() + +# ------------------------------ +# PUBLIC METHODS +# ------------------------------ + +def cleanup_infra_deployments(deployment: INFRASTRUCTURE, indexes: int | list[int] | None = None) -> None: + """ + Clean up infrastructure deployments by deployment enum and index/indexes. + Obtains the infra resource group name for each index and calls the private cleanup method. + + Args: + deployment (INFRASTRUCTURE): The infrastructure deployment enum value. + indexes (int | list[int] | None): A single index, a list of indexes, or None for no index. + """ + validate_infrastructure(deployment) + + if indexes is None: + indexes_list = [None] + elif isinstance(indexes, (list, tuple)): + indexes_list = list(indexes) + else: + indexes_list = [indexes] + + for idx in indexes_list: + print_info(f"Cleaning up resources for {deployment} - {idx}", True) + rg_name = get_infra_rg_name(deployment, idx) + _cleanup_resources(deployment.value, rg_name) + +def cleanup_deployment(deployment: str, indexes: int | list[int] | None = None) -> None: + """ + Clean up sample deployments by deployment name and index/indexes. + Obtains the resource group name for each index and calls the private cleanup method. + + Args: + deployment (str): The deployment name (string). + indexes (int | list[int] | None): A single index, a list of indexes, or None for no index. + """ + if not isinstance(deployment, str): + raise ValueError("deployment must be a string") + if indexes is None: + indexes_list = [None] + elif isinstance(indexes, (list, tuple)): + indexes_list = list(indexes) + else: + indexes_list = [indexes] + for idx in indexes_list: + rg_name = get_rg_name(deployment, idx) + _cleanup_resources(deployment, rg_name) + def extract_json(text: str) -> any: """ - Extract the first valid JSON object or array from a string and return it as a JSON string. - Uses json.JSONDecoder().raw_decode to robustly parse the first JSON object or array found in the string. The JSON object may be buried in the string and have preceding or trailing regular text. + Extract the first valid JSON object or array from a string and return it as a Python object. + + This function searches the input string for the first occurrence of a JSON object or array (delimited by '{' or '['), + and attempts to decode it using json.JSONDecoder().raw_decode. If the input is already valid JSON, it is returned as a Python object. + If no valid JSON is found, None is returned. Args: - text (str): The string to search for JSON. + text (str): The string to search for a JSON object or array. Returns: - str | None: The extracted JSON as a string, or None if not found. + Any | None: The extracted JSON as a Python object (dict or list), or None if not found or not valid. """ if not isinstance(text, str): return None - - # If the string is already JSON, return it. - if is_string_json(text): - return text - print(text) + # If the string is already valid JSON, parse and return it as a Python object. + if is_string_json(text): + return json.loads(text) decoder = json.JSONDecoder() for start in range(len(text)): if text[start] in ('{', '['): try: - obj, end = decoder.raw_decode(text[start:]) - return json.loads(obj) + obj, _ = decoder.raw_decode(text[start:]) + return obj except Exception: continue @@ -380,10 +423,14 @@ def is_string_json(text: str) -> bool: bool: True if the string is valid JSON, False otherwise. """ + # Accept only str, bytes, or bytearray as valid input for JSON parsing. + if not isinstance(text, (str, bytes, bytearray)): + return False + try: json.loads(text) return True - except ValueError: + except (ValueError, TypeError): return False def get_account_info() -> Tuple[str, str, str]: @@ -538,11 +585,13 @@ def run(command: str, ok_message: str = '', error_message: str = '', print_outpu start_time = time.time() # Execute the command and capture the output + try: output_text = subprocess.check_output(command, shell = True, stderr = subprocess.STDOUT).decode("utf-8") success = True - except subprocess.CalledProcessError as e: - output_text = e.output.decode("utf-8") + except Exception as e: + # Handles both CalledProcessError and any custom/other exceptions (for test mocks) + output_text = getattr(e, 'output', b'').decode("utf-8") if hasattr(e, 'output') and isinstance(e.output, (bytes, bytearray)) else str(e) success = False if print_output: diff --git a/tests/python/.coveragerc b/tests/python/.coveragerc new file mode 100644 index 0000000..a7cf676 --- /dev/null +++ b/tests/python/.coveragerc @@ -0,0 +1,11 @@ +[run] +branch = True +source = + shared/python +omit = + */__init__.py + */__pycache__/* + +[report] +show_missing = True +skip_covered = True diff --git a/shared/python-tests/.pylintrc b/tests/python/.pylintrc similarity index 68% rename from shared/python-tests/.pylintrc rename to tests/python/.pylintrc index 6fda18c..6bada50 100644 --- a/shared/python-tests/.pylintrc +++ b/tests/python/.pylintrc @@ -5,3 +5,6 @@ disable = C0115, # Missing class docstring C0116, # Missing function or method docstring W0212, # Access to a protected member _ of a client class + R0903, # Too few public methods + R0913, # Too many arguments + W0621, # Redefining name from outer scope diff --git a/tests/python/__init__.py b/tests/python/__init__.py new file mode 100644 index 0000000..d99e668 --- /dev/null +++ b/tests/python/__init__.py @@ -0,0 +1,3 @@ +""" +Unit tests package for API Management samples. +""" diff --git a/tests/python/conftest.py b/tests/python/conftest.py new file mode 100644 index 0000000..a0ecccc --- /dev/null +++ b/tests/python/conftest.py @@ -0,0 +1,36 @@ +""" +Shared test configuration and fixtures for pytest. +""" +import os +import sys +from typing import Any + +import pytest + +# Add the shared/python directory to the Python path for all tests +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../shared/python'))) + + +# ------------------------------ +# SHARED FIXTURES +# ------------------------------ + +@pytest.fixture(scope="session") +def shared_python_path() -> str: + """Provide the path to the shared Python modules.""" + return os.path.abspath(os.path.join(os.path.dirname(__file__), '../../shared/python')) + +@pytest.fixture(scope="session") +def test_data_path() -> str: + """Provide the path to test data files.""" + return os.path.abspath(os.path.join(os.path.dirname(__file__), 'data')) + +@pytest.fixture +def sample_test_data() -> dict[str, Any]: + """Provide sample test data for use across tests.""" + return { + "test_url": "https://test-apim.azure-api.net", + "test_subscription_key": "test-subscription-key-12345", + "test_resource_group": "rg-test-apim-01", + "test_location": "eastus2" + } diff --git a/tests/python/pytest.ini b/tests/python/pytest.ini new file mode 100644 index 0000000..5293117 --- /dev/null +++ b/tests/python/pytest.ini @@ -0,0 +1,13 @@ +[pytest] +markers = + apimtypes: tests for apimtypes module + apimrequests: tests for apimrequests module + utils: tests for utils module + integration: integration tests that require external services + slow: tests that take a long time to run + unit: marks tests as unit tests + http: marks tests that mock or use HTTP +testpaths = + tests/python +python_files = + test_*.py diff --git a/tests/python/run_tests.ps1 b/tests/python/run_tests.ps1 new file mode 100644 index 0000000..a5cd748 --- /dev/null +++ b/tests/python/run_tests.ps1 @@ -0,0 +1,3 @@ +# PowerShell script to run pytest with coverage and store .coverage in tests/python +$env:COVERAGE_FILE = "tests/python/.coverage" +pytest -v --cov=shared/python --cov-config=tests/python/.coveragerc --cov-report=html:tests/python/htmlcov tests/python/ diff --git a/tests/python/run_tests.sh b/tests/python/run_tests.sh new file mode 100644 index 0000000..b5160e9 --- /dev/null +++ b/tests/python/run_tests.sh @@ -0,0 +1,4 @@ +# Shell script to run pytest with coverage and store .coverage in tests/python +COVERAGE_FILE=tests/python/.coverage +export COVERAGE_FILE +pytest -v --cov=shared/python --cov-config=tests/python/.coveragerc --cov-report=html:tests/python/htmlcov tests/python/ diff --git a/tests/python/test_apimrequests.py b/tests/python/test_apimrequests.py new file mode 100644 index 0000000..4ce57fd --- /dev/null +++ b/tests/python/test_apimrequests.py @@ -0,0 +1,194 @@ +import pytest +import requests +from unittest.mock import patch, MagicMock +from shared.python.apimrequests import ApimRequests +from shared.python.apimtypes import HTTP_VERB, SUBSCRIPTION_KEY_PARAMETER_NAME + +# Sample values for tests +default_url = "https://example.com/apim/" +default_key = "test-key" +default_path = "/test" +default_headers = {"Custom-Header": "Value"} +default_data = {"foo": "bar"} + +@pytest.fixture +def apim(): + return ApimRequests(default_url, default_key) + + +@pytest.mark.unit +def test_init_sets_headers(): + """Test that headers are set correctly when subscription key is provided.""" + apim = ApimRequests(default_url, default_key) + assert apim.url == default_url + assert apim.apimSubscriptionKey == default_key + assert apim.headers[SUBSCRIPTION_KEY_PARAMETER_NAME] == default_key + + +@pytest.mark.unit +def test_init_no_key(): + """Test that headers are set correctly when no subscription key is provided.""" + apim = ApimRequests(default_url) + assert apim.url == default_url + assert apim.apimSubscriptionKey is None + assert "Ocp-Apim-Subscription-Key" not in apim.headers + assert apim.headers["Accept"] == "application/json" + +@pytest.mark.http +@patch("shared.python.apimrequests.requests.request") +@patch("shared.python.apimrequests.utils.print_message") +@patch("shared.python.apimrequests.utils.print_info") +@patch("shared.python.apimrequests.utils.print_error") +def test_single_get_success(mock_print_error, mock_print_info, mock_print_message, mock_request, apim): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"Content-Type": "application/json"} + mock_response.json.return_value = {"result": "ok"} + mock_response.text = '{"result": "ok"}' + mock_response.raise_for_status.return_value = None + mock_request.return_value = mock_response + + with patch.object(apim, "_print_response") as mock_print_response: + result = apim.singleGet(default_path, printResponse=True) + assert result == '{\n "result": "ok"\n}' + mock_print_response.assert_called_once_with(mock_response) + mock_print_error.assert_not_called() + +@pytest.mark.http +@patch("shared.python.apimrequests.requests.request") +@patch("shared.python.apimrequests.utils.print_message") +@patch("shared.python.apimrequests.utils.print_info") +@patch("shared.python.apimrequests.utils.print_error") +def test_single_get_error(mock_print_error, mock_print_info, mock_print_message, mock_request, apim): + mock_request.side_effect = requests.exceptions.RequestException("fail") + result = apim.singleGet(default_path, printResponse=True) + assert result is None + mock_print_error.assert_called_once() + +@pytest.mark.http +@patch("shared.python.apimrequests.requests.request") +@patch("shared.python.apimrequests.utils.print_message") +@patch("shared.python.apimrequests.utils.print_info") +@patch("shared.python.apimrequests.utils.print_error") +def test_single_post_success(mock_print_error, mock_print_info, mock_print_message, mock_request, apim): + mock_response = MagicMock() + mock_response.status_code = 201 + mock_response.headers = {"Content-Type": "application/json"} + mock_response.json.return_value = {"created": True} + mock_response.text = '{"created": true}' + mock_response.raise_for_status.return_value = None + mock_request.return_value = mock_response + + with patch.object(apim, "_print_response") as mock_print_response: + result = apim.singlePost(default_path, data=default_data, printResponse=True) + assert result == '{\n "created": true\n}' + mock_print_response.assert_called_once_with(mock_response) + mock_print_error.assert_not_called() + +@pytest.mark.http +@patch("shared.python.apimrequests.requests.Session") +@patch("shared.python.apimrequests.utils.print_message") +@patch("shared.python.apimrequests.utils.print_info") +def test_multi_get_success(mock_print_info, mock_print_message, mock_session, apim): + mock_sess = MagicMock() + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"Content-Type": "application/json"} + mock_response.json.return_value = {"result": "ok"} + mock_response.text = '{"result": "ok"}' + mock_response.raise_for_status.return_value = None + mock_sess.request.return_value = mock_response + mock_session.return_value = mock_sess + + with patch.object(apim, "_print_response_code") as mock_print_code: + result = apim.multiGet(default_path, runs=2, printResponse=True) + assert len(result) == 2 + for run in result: + assert run["status_code"] == 200 + assert run["response"] == '{\n "result": "ok"\n}' + assert mock_sess.request.call_count == 2 + mock_print_code.assert_called() + +@pytest.mark.http +@patch("shared.python.apimrequests.requests.Session") +@patch("shared.python.apimrequests.utils.print_message") +@patch("shared.python.apimrequests.utils.print_info") +def test_multi_get_error(mock_print_info, mock_print_message, mock_session, apim): + mock_sess = MagicMock() + mock_sess.request.side_effect = requests.exceptions.RequestException("fail") + mock_session.return_value = mock_sess + with patch.object(apim, "_print_response_code"): + # Should raise inside the loop and propagate the exception, ensuring the session is closed + with pytest.raises(requests.exceptions.RequestException): + apim.multiGet(default_path, runs=1, printResponse=True) + + +# Sample values for tests +url = "https://example.com/apim/" +key = "test-key" +path = "/test" + +def make_apim(): + return ApimRequests(url, key) + +@pytest.mark.http +def test_single_post_error(): + apim = make_apim() + with patch("shared.python.apimrequests.requests.request") as mock_request, \ + patch("shared.python.apimrequests.utils.print_error") as mock_print_error: + import requests + mock_request.side_effect = requests.RequestException("fail") + result = apim.singlePost(path, data={"foo": "bar"}, printResponse=True) + assert result is None + mock_print_error.assert_called() + +@pytest.mark.http +def test_multi_get_non_json(): + apim = make_apim() + with patch("shared.python.apimrequests.requests.Session") as mock_session: + mock_sess = MagicMock() + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"Content-Type": "text/plain"} + mock_response.text = "not json" + mock_response.raise_for_status.return_value = None + mock_sess.request.return_value = mock_response + mock_session.return_value = mock_sess + with patch.object(apim, "_print_response_code"): + result = apim.multiGet(path, runs=1, printResponse=True) + assert result[0]["response"] == "not json" + +@pytest.mark.http +def test_request_header_merging(): + apim = make_apim() + with patch("shared.python.apimrequests.requests.request") as mock_request: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.headers = {"Content-Type": "application/json"} + mock_response.json.return_value = {"ok": True} + mock_response.text = '{"ok": true}' + mock_response.raise_for_status.return_value = None + mock_request.return_value = mock_response + # Custom header should override default + custom_headers = {"Accept": "application/xml", "X-Test": "1"} + with patch.object(apim, "_print_response"): + apim.singleGet(path, headers=custom_headers, printResponse=True) + called_headers = mock_request.call_args[1]["headers"] + assert called_headers["Accept"] == "application/xml" + assert called_headers["X-Test"] == "1" + +@pytest.mark.http +def test_init_missing_url(): + # Negative: missing URL should raise TypeError + with pytest.raises(TypeError): + ApimRequests() + +@pytest.mark.http +def test_print_response_code_edge(): + apim = make_apim() + class DummyResponse: + status_code = 302 + reason = "Found" + with patch("shared.python.apimrequests.utils.print_val") as mock_print_val: + apim._print_response_code(DummyResponse()) + mock_print_val.assert_called_with("Response status", "302") diff --git a/tests/python/test_apimtypes.py b/tests/python/test_apimtypes.py new file mode 100644 index 0000000..33c4a49 --- /dev/null +++ b/tests/python/test_apimtypes.py @@ -0,0 +1,241 @@ +""" +Unit tests for apimtypes.py. +""" +import pytest +from shared.python import apimtypes + + +# ------------------------------ +# CONSTANTS +# ------------------------------ + +EXAMPLE_NAME = "test-api" +EXAMPLE_DISPLAY_NAME = "Test API" +EXAMPLE_PATH = "/test" +EXAMPLE_DESCRIPTION = "A test API." +EXAMPLE_POLICY_XML = "" + + +# ------------------------------ +# TEST METHODS +# ------------------------------ + +@pytest.mark.unit +def test_api_creation(): + """Test creation of API object and its attributes.""" + api = apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML, + operations = None + ) + + assert api.name == EXAMPLE_NAME + assert api.displayName == EXAMPLE_DISPLAY_NAME + assert api.path == EXAMPLE_PATH + assert api.description == EXAMPLE_DESCRIPTION + assert api.policyXml == EXAMPLE_POLICY_XML + assert api.operations == [] + + +@pytest.mark.unit +def test_api_repr(): + """Test __repr__ method of API.""" + api = apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML, + operations = None + ) + result = repr(api) + assert "API" in result + assert EXAMPLE_NAME in result + assert EXAMPLE_DISPLAY_NAME in result + +@pytest.mark.unit +def test_api_equality(): + """Test equality comparison for API objects. + """ + api1 = apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML, + operations = None + ) + api2 = apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML, + operations = None + ) + assert api1 == api2 + +def test_api_inequality(): + """ + Test inequality for API objects with different attributes. + """ + api1 = apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML, + operations = None + ) + api2 = apimtypes.API( + name = "other-api", + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML, + operations = None + ) + assert api1 != api2 + +def test_api_missing_fields(): + """ + Test that missing required fields raise TypeError. + """ + with pytest.raises(TypeError): + apimtypes.API( + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML + ) + + with pytest.raises(TypeError): + apimtypes.API( + name = EXAMPLE_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML + ) + + with pytest.raises(TypeError): + apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + description = EXAMPLE_DESCRIPTION, + policyXml = EXAMPLE_POLICY_XML + ) + + with pytest.raises(TypeError): + apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + policyXml = EXAMPLE_POLICY_XML + ) + + with pytest.raises(TypeError): + apimtypes.API( + name = EXAMPLE_NAME, + displayName = EXAMPLE_DISPLAY_NAME, + path = EXAMPLE_PATH, + description = EXAMPLE_DESCRIPTION + ) + + +# ------------------------------ +# ENUMS +# ------------------------------ + +def test_apimnetworkmode_enum(): + assert apimtypes.APIMNetworkMode.PUBLIC == "Public" + assert apimtypes.APIMNetworkMode.EXTERNAL_VNET == "External" + assert apimtypes.APIMNetworkMode.INTERNAL_VNET == "Internal" + assert apimtypes.APIMNetworkMode.NONE == "None" + with pytest.raises(ValueError): + apimtypes.APIMNetworkMode("invalid") + +def test_apim_sku_enum(): + assert apimtypes.APIM_SKU.DEVELOPER == "Developer" + assert apimtypes.APIM_SKU.BASIC == "Basic" + assert apimtypes.APIM_SKU.STANDARD == "Standard" + assert apimtypes.APIM_SKU.PREMIUM == "Premium" + assert apimtypes.APIM_SKU.BASICV2 == "Basicv2" + assert apimtypes.APIM_SKU.STANDARDV2 == "Standardv2" + assert apimtypes.APIM_SKU.PREMIUMV2 == "Premiumv2" + with pytest.raises(ValueError): + apimtypes.APIM_SKU("invalid") + +def test_http_verb_enum(): + assert apimtypes.HTTP_VERB.GET == "GET" + assert apimtypes.HTTP_VERB.POST == "POST" + assert apimtypes.HTTP_VERB.PUT == "PUT" + assert apimtypes.HTTP_VERB.DELETE == "DELETE" + assert apimtypes.HTTP_VERB.PATCH == "PATCH" + assert apimtypes.HTTP_VERB.OPTIONS == "OPTIONS" + assert apimtypes.HTTP_VERB.HEAD == "HEAD" + with pytest.raises(ValueError): + apimtypes.HTTP_VERB("FOO") + +def test_infrastructure_enum(): + assert apimtypes.INFRASTRUCTURE.SIMPLE_APIM == "simple-apim" + assert apimtypes.INFRASTRUCTURE.APIM_ACA == "apim-aca" + assert apimtypes.INFRASTRUCTURE.AFD_APIM_PE == "afd-apim-pe" + with pytest.raises(ValueError): + apimtypes.INFRASTRUCTURE("bad") + + +# ------------------------------ +# OPERATION CLASSES +# ------------------------------ + +def test_apioperation_to_dict(): + op = apimtypes.APIOperation( + name="op1", + displayName="Operation 1", + urlTemplate="/foo", + method=apimtypes.HTTP_VERB.GET, + description="desc", + policyXml="" + ) + d = op.to_dict() + assert d["name"] == "op1" + assert d["displayName"] == "Operation 1" + assert d["urlTemplate"] == "/foo" + assert d["method"] == apimtypes.HTTP_VERB.GET + assert d["description"] == "desc" + assert d["policyXml"] == "" + +def test_get_apioperation(): + op = apimtypes.GET_APIOperation(description="desc", policyXml="") + assert op.name == "GET" + assert op.method == apimtypes.HTTP_VERB.GET + assert op.urlTemplate == "/" + assert op.description == "desc" + assert op.policyXml == "" + d = op.to_dict() + assert d["method"] == apimtypes.HTTP_VERB.GET + +def test_post_apioperation(): + op = apimtypes.POST_APIOperation(description="desc", policyXml="") + assert op.name == "POST" + assert op.method == apimtypes.HTTP_VERB.POST + assert op.urlTemplate == "/" + assert op.description == "desc" + assert op.policyXml == "" + d = op.to_dict() + assert d["method"] == apimtypes.HTTP_VERB.POST + +def test_apioperation_invalid_method(): + # Negative: method must be a valid HTTP_VERB + with pytest.raises(ValueError): + apimtypes.APIOperation( + name="bad", + displayName="Bad", + urlTemplate="/bad", + method="INVALID", + description="desc", + policyXml="" + ) diff --git a/tests/python/test_utils.py b/tests/python/test_utils.py new file mode 100644 index 0000000..3e49eab --- /dev/null +++ b/tests/python/test_utils.py @@ -0,0 +1,226 @@ +import os +import builtins +import pytest +from unittest.mock import patch, MagicMock, mock_open +from shared.python import utils +from apimtypes import INFRASTRUCTURE + +# ------------------------------ +# is_string_json +# ------------------------------ + +@pytest.mark.parametrize( + "input_str,expected", + [ + ("{\"a\": 1}", True), + ("[1, 2, 3]", True), + ("not json", False), + ("{\"a\": 1", False), + ("", False), + (None, False), + (123, False), + ] +) +def test_is_string_json(input_str, expected): + assert utils.is_string_json(input_str) is expected + +# ------------------------------ +# get_account_info +# ------------------------------ + +def test_get_account_info_success(monkeypatch): + mock_json = { + 'user': {'name': 'testuser'}, + 'tenantId': 'tenant', + 'id': 'subid' + } + mock_output = MagicMock(success=True, json_data=mock_json) + monkeypatch.setattr(utils, 'run', lambda *a, **kw: mock_output) + result = utils.get_account_info() + assert result == ('testuser', 'tenant', 'subid') + +def test_get_account_info_failure(monkeypatch): + mock_output = MagicMock(success=False, json_data=None) + monkeypatch.setattr(utils, 'run', lambda *a, **kw: mock_output) + with pytest.raises(Exception): + utils.get_account_info() + +# ------------------------------ +# get_deployment_name +# ------------------------------ + +def test_get_deployment_name(monkeypatch): + monkeypatch.setattr(os, 'getcwd', lambda: '/foo/bar/baz') + assert utils.get_deployment_name() == 'baz' + +def test_get_deployment_name_error(monkeypatch): + monkeypatch.setattr(os, 'getcwd', lambda: '') + with pytest.raises(RuntimeError): + utils.get_deployment_name() + +# ------------------------------ +# get_frontdoor_url +# ------------------------------ + +def test_get_frontdoor_url_success(monkeypatch): + mock_profile = [{"name": "afd1"}] + mock_endpoints = [{"hostName": "foo.azurefd.net"}] + def run_side_effect(cmd, *a, **kw): + if 'profile list' in cmd: + return MagicMock(success=True, json_data=mock_profile) + if 'endpoint list' in cmd: + return MagicMock(success=True, json_data=mock_endpoints) + return MagicMock(success=False, json_data=None) + monkeypatch.setattr(utils, 'run', run_side_effect) + url = utils.get_frontdoor_url(INFRASTRUCTURE.AFD_APIM_PE, 'rg') + assert url == 'https://foo.azurefd.net' + +def test_get_frontdoor_url_none(monkeypatch): + monkeypatch.setattr(utils, 'run', lambda *a, **kw: MagicMock(success=False, json_data=None)) + url = utils.get_frontdoor_url(INFRASTRUCTURE.AFD_APIM_PE, 'rg') + assert url is None + +# ------------------------------ +# get_infra_rg_name & get_rg_name +# ------------------------------ + +def test_get_infra_rg_name(monkeypatch): + class DummyInfra: + value = 'foo' + monkeypatch.setattr(utils, 'validate_infrastructure', lambda x: x) + assert utils.get_infra_rg_name(DummyInfra) == 'apim-infra-foo' + assert utils.get_infra_rg_name(DummyInfra, 2) == 'apim-infra-foo-2' + +def test_get_rg_name(): + assert utils.get_rg_name('foo') == 'apim-sample-foo' + assert utils.get_rg_name('foo', 3) == 'apim-sample-foo-3' + +# ------------------------------ +# run +# ------------------------------ + +def test_run_success(monkeypatch): + monkeypatch.setattr('subprocess.check_output', lambda *a, **kw: b'{"a": 1}') + out = utils.run('echo', print_command_to_run=False) + assert out.success is True + assert out.json_data == {"a": 1} + +def test_run_failure(monkeypatch): + class DummyErr(Exception): + output = b'fail' + def fail(*a, **kw): + raise DummyErr() + monkeypatch.setattr('subprocess.check_output', fail) + out = utils.run('bad', print_command_to_run=False) + assert out.success is False + assert isinstance(out.text, str) + +# ------------------------------ +# create_resource_group & does_resource_group_exist +# ------------------------------ + +def test_does_resource_group_exist(monkeypatch): + monkeypatch.setattr(utils, 'run', lambda *a, **kw: MagicMock(success=True)) + assert utils.does_resource_group_exist('foo') is True + monkeypatch.setattr(utils, 'run', lambda *a, **kw: MagicMock(success=False)) + assert utils.does_resource_group_exist('foo') is False + +def test_create_resource_group(monkeypatch): + called = {} + monkeypatch.setattr(utils, 'does_resource_group_exist', lambda rg: False) + monkeypatch.setattr(utils, 'print_info', lambda *a, **kw: called.setdefault('info', True)) + monkeypatch.setattr(utils, 'run', lambda *a, **kw: called.setdefault('run', True)) + utils.create_resource_group('foo', 'bar') + assert called['info'] and called['run'] + +# ------------------------------ +# policy_xml_replacement +# ------------------------------ + +def test_policy_xml_replacement(monkeypatch): + m = mock_open(read_data='foo') + monkeypatch.setattr(builtins, 'open', m) + assert utils.policy_xml_replacement('dummy.xml') == 'foo' + +# ------------------------------ +# cleanup_resources (smoke) +# ------------------------------ + +def test_cleanup_resources_smoke(monkeypatch): + monkeypatch.setattr(utils, 'run', lambda *a, **kw: MagicMock(success=True, json_data={})) + monkeypatch.setattr(utils, 'print_info', lambda *a, **kw: None) + monkeypatch.setattr(utils, 'print_error', lambda *a, **kw: None) + monkeypatch.setattr(utils, 'print_message', lambda *a, **kw: None) + monkeypatch.setattr(utils, 'print_ok', lambda *a, **kw: None) + monkeypatch.setattr(utils, 'print_warning', lambda *a, **kw: None) + monkeypatch.setattr(utils, 'print_val', lambda *a, **kw: None) + # Direct private method call for legacy test (should still work) + utils._cleanup_resources(INFRASTRUCTURE.SIMPLE_APIM.value, 'rg') + +def test_cleanup_infra_deployment_single(monkeypatch): + monkeypatch.setattr(utils, '_cleanup_resources', lambda deployment_name, rg_name: None) + utils.cleanup_infra_deployments(INFRASTRUCTURE.SIMPLE_APIM, None) + utils.cleanup_infra_deployments(INFRASTRUCTURE.SIMPLE_APIM, 1) + utils.cleanup_infra_deployments(INFRASTRUCTURE.SIMPLE_APIM, [1, 2]) + +def test_cleanup_deployment_single(monkeypatch): + monkeypatch.setattr(utils, '_cleanup_resources', lambda deployment_name, rg_name: None) + utils.cleanup_deployment('foo', None) + utils.cleanup_deployment('foo', 1) + utils.cleanup_deployment('foo', [1, 2]) + +# ------------------------------ +# EXTRACT_JSON EDGE CASES +# ------------------------------ + +@pytest.mark.parametrize( + "input_val,expected", + [ + (None, None), + (123, None), + ([], None), + ("", None), + (" ", None), + ("not json", None), + ("{\"a\": 1}", {"a": 1}), + ("[1, 2, 3]", [1, 2, 3]), + (" {\"a\": 1} ", {"a": 1}), + ("prefix {\"foo\": 42} suffix", {"foo": 42}), + ("prefix [1, 2, 3] suffix", [1, 2, 3]), + ("{\"a\": 1}{\"b\": 2}", {"a": 1}), # Only first JSON object + ("[1, 2, 3][4, 5, 6]", [1, 2, 3]), # Only first JSON array + ("{\"a\": 1,}", None), # Trailing comma + ("[1, 2,]", None), # Trailing comma in array + ("{\"a\": [1, 2, {\"b\": 3}]}", {"a": [1, 2, {"b": 3}]}), + ("\n\t{\"a\": 1}\n", {"a": 1}), + ("{\"a\": \"b \\u1234\"}", {"a": "b \u1234"}), + ("{\"a\": 1} [2, 3]", {"a": 1}), # Object before array + ("[2, 3] {\"a\": 1}", [2, 3]), # Array before object + ("{\"a\": 1, \"b\": {\"c\": 2}}", {"a": 1, "b": {"c": 2}}), + ("{\"a\": 1, \"b\": [1, 2, 3]}", {"a": 1, "b": [1, 2, 3]}), + ("\n\n[\n1, 2, 3\n]\n", [1, 2, 3]), + ("{\"a\": 1, \"b\": null}", {"a": 1, "b": None}), + ("{\"a\": true, \"b\": false}", {"a": True, "b": False}), + ("{\"a\": 1, \"b\": \"c\"}", {"a": 1, "b": "c"}), + ("{\"a\": 1, \"b\": [1, 2, {\"c\": 3}]} ", {"a": 1, "b": [1, 2, {"c": 3}]}), + ("{\"a\": 1, \"b\": [1, 2, {\"c\": 3, \"d\": [4, 5]}]} ", {"a": 1, "b": [1, 2, {"c": 3, "d": [4, 5]}]}), + ] +) +def test_extract_json_edge_cases(input_val, expected): + """Test extract_json with a wide range of edge cases and malformed input.""" + result = utils.extract_json(input_val) + assert result == expected + +def test_extract_json_large_object(): + """Test extract_json with a large JSON object.""" + large_obj = {"a": list(range(1000)), "b": {"c": "x" * 1000}} + import json + s = json.dumps(large_obj) + assert utils.extract_json(s) == large_obj + +def test_extract_json_multiple_json_types(): + """Test extract_json returns the first valid JSON (object or array) in the string.""" + s = '[1,2,3]{"a": 1}' + assert utils.extract_json(s) == [1, 2, 3] + s2 = '{"a": 1}[1,2,3]' + assert utils.extract_json(s2) == {"a": 1}