import pytest import requests import requests_mock from app.clients.performance_platform.performance_platform_client import ( PerformancePlatformClient, ) @pytest.fixture(scope="function") def perf_client(client, mocker): perf_client = PerformancePlatformClient() current_app = mocker.Mock( config={ "PERFORMANCE_PLATFORM_ENABLED": True, "PERFORMANCE_PLATFORM_ENDPOINTS": {"foo": "my_token", "bar": "other_token"}, "PERFORMANCE_PLATFORM_URL": "https://performance-platform-url/", } ) perf_client.init_app(current_app) return perf_client def test_should_not_call_if_not_enabled(perf_client): with requests_mock.Mocker() as request_mock: request_mock.post( "https://performance-platform-url/foo", json={}, status_code=200 ) perf_client._active = False perf_client.send_stats_to_performance_platform({"dataType": "foo"}) assert request_mock.called is False def test_should_call_datatype_endpoint_if_enabled(perf_client): with requests_mock.Mocker() as request_mock: request_mock.post( "https://performance-platform-url/foo", json={}, status_code=200 ) perf_client.send_stats_to_performance_platform({"dataType": "foo"}) assert request_mock.call_count == 1 assert request_mock.last_request.method == "POST" @pytest.mark.parametrize( "dataset, token", [("foo", "my_token"), ("bar", "other_token")] ) def test_should_use_correct_token(perf_client, dataset, token): with requests_mock.Mocker() as request_mock: request_mock.post( "https://performance-platform-url/foo", json={}, status_code=200 ) request_mock.post( "https://performance-platform-url/bar", json={}, status_code=200 ) perf_client.send_stats_to_performance_platform({"dataType": dataset}) assert request_mock.call_count == 1 assert request_mock.last_request.headers.get("authorization") == "Bearer {}".format( token ) def test_should_raise_for_status(perf_client): with pytest.raises(requests.HTTPError), requests_mock.Mocker() as request_mock: request_mock.post( "https://performance-platform-url/foo", json={}, status_code=403 ) perf_client.send_stats_to_performance_platform({"dataType": "foo"}) def test_generate_payload_id(): payload = { "_timestamp": "2023-01-01 00:00:00", "service": "my_service", "group_name": "group_name", "dataType": "dataType", "period": "period", } result = PerformancePlatformClient.generate_payload_id(payload, "group_name") assert ( result == "MjAyMy0wMS0wMSAwMDowMDowMG15X3NlcnZpY2Vncm91cF9uYW1lZGF0YVR5cGVwZXJpb2Q=" )