import pytest import requests from datadog_tools.connectors.sonar import _MAX_TRIES, Sonar @pytest.fixture def mock_get_project_request_exists(mocker): mock_response = {"components": [{"key": "test-service", "name": "Test Service"}]} mocker.patch( "datadog_tools.connectors.sonar.requests.get", return_value=mocker.Mock(status_code=200, json=lambda: mock_response, raise_for_status=lambda: None), ) @pytest.fixture def mock_get_project_request_not_exists(mocker): mocker.patch( "datadog_tools.connectors.sonar.requests.get", return_value=mocker.Mock(status_code=200, json=lambda: {}, raise_for_status=lambda: None), ) def test_get_project_found(mock_get_project_request_exists): sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") project = sonar.get_project("test-service") assert project == {"key": "test-service", "name": "Test Service"} def test_get_project_not_found(mock_get_project_request_not_exists): sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") project = sonar.get_project("test-service") assert project is None def test_get_project_not_found_does_not_retry(mocker): mock_get = mocker.patch( "datadog_tools.connectors.sonar.requests.get", return_value=mocker.Mock(status_code=200, json=lambda: {}, raise_for_status=lambda: None), ) sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") sonar.get_project("test-service") mock_get.assert_called_once() @pytest.mark.parametrize( "exc_class", [ requests.exceptions.ReadTimeout, requests.exceptions.ConnectionError, requests.exceptions.HTTPError, ], ) def test_get_project_connection_error_returns_none_after_retries(mocker, exc_class): mocker.patch( "datadog_tools.connectors.sonar.requests.get", side_effect=exc_class("simulated error"), ) mocker.patch("time.sleep") sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") project = sonar.get_project("test-service") assert project is None def test_get_project_retries_before_giving_up(mocker): mocker.patch( "datadog_tools.connectors.sonar.requests.get", side_effect=requests.exceptions.ReadTimeout("timed out"), ) mocker.patch("time.sleep") sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") sonar.get_project("test-service") assert requests.get.call_count == _MAX_TRIES def test_get_project_succeeds_on_retry(mocker): mock_response = {"components": [{"key": "test-service", "name": "Test Service"}]} mocker.patch( "datadog_tools.connectors.sonar.requests.get", side_effect=[ requests.exceptions.ReadTimeout("timed out"), mocker.Mock(status_code=200, json=lambda: mock_response, raise_for_status=lambda: None), ], ) mocker.patch("time.sleep") sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") project = sonar.get_project("test-service") assert project == {"key": "test-service", "name": "Test Service"} def test_get_project_giveup_prints_warning(mocker, capsys): mocker.patch( "datadog_tools.connectors.sonar.requests.get", side_effect=requests.exceptions.ReadTimeout("timed out"), ) mocker.patch("time.sleep") sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") sonar.get_project("test-service") output = capsys.readouterr().out assert "Warning" in output assert "test-service" in output def test_get_project_url(): sonar = Sonar(base_url="https://sonar.example.com", api_token="test-token") url = sonar.get_project_url("test-service") assert url == "https://sonar.example.com/dashboard?id=test-service"