"""Test Auth0 Connectors.""" from typing import Any from unittest.mock import MagicMock, call, patch from uuid import UUID import pytest from auth0.exceptions import Auth0Error from auth0.management import ClientGrants as ClientGrantsAPI, Clients as MgmtClientsAPI from pydantic import ValidationError from m2mconfig import config, constants from m2mconfig.connectors.auth0_connector import ( GET_CLIENT_EXTRA_PARAMS, GET_CLIENT_FIELDS, Auth0ClientConfiguration, Auth0ClientResponse, Auth0Connector, ClientGrantsCreateError, clients_dict_by_name, get_auth0_mgmt_client_api, get_client_grants_api, get_mgmt_api_token, ) from m2mconfig.schemas import ( ClientGrants, ClientGrantsValidator, ClientMetadata, RegistryEntry, ) @pytest.fixture() def mock_client_grants_api() -> ClientGrantsAPI: """Mock a MgmtClientsAPI instance.""" return MagicMock(spec=ClientGrantsAPI) @pytest.fixture() def mock_auth0_mgmt_client_api() -> MgmtClientsAPI: """Auth0 client api mock.""" return MagicMock(spec=MgmtClientsAPI) @patch("m2mconfig.connectors.auth0_connector.MgmtClientsAPI") def test_get_auth0_mgmt_client_api( mock_mgmt_clients_api: MagicMock, monkeypatch: pytest.MonkeyPatch ) -> None: """Test the get_auth0_mgmt_client_api factory function.""" monkeypatch.setattr(config, "AUTH0_DOMAIN", "fake.domain") mock_mgmt_clients_api_inst = mock_mgmt_clients_api.return_value ret = get_auth0_mgmt_client_api(mgmt_api_token="test.token") assert ret == mock_mgmt_clients_api_inst mock_mgmt_clients_api.assert_called_with("fake.domain", "test.token") @patch("m2mconfig.connectors.auth0_connector.ClientGrantsAPI") def test_get_client_grants_api( mock_client_grants_api: MagicMock, monkeypatch: pytest.MonkeyPatch ) -> None: """Test the get_client_grants_api factory function.""" monkeypatch.setattr(config, "AUTH0_DOMAIN", "fake.domain") mock_client_grants_api_inst = mock_client_grants_api.return_value ret = get_client_grants_api(mgmt_api_token="test.token") assert ret == mock_client_grants_api_inst mock_client_grants_api.assert_called_with("fake.domain", "test.token") @patch("m2mconfig.connectors.auth0_connector.GetToken") def test_get_mgmt_api_token( mock_get_token: MagicMock, monkeypatch: pytest.MonkeyPatch ) -> None: """Test the get_mgmt_api_token function.""" monkeypatch.setattr(config, "AUTH0_DOMAIN", "fake.domain") monkeypatch.setattr( config, "AUTH0_M2M_CONFIG_CLIENT_CREDENTIALS", '{"audience":"https://dev-orchard.auth0.com/api/v2/","client_id":"test_client_id","client_secret":"test_client_secret","grant_type":"client_credentials"}', # noqa: E501 ) mock_get_token_inst = mock_get_token.return_value mock_get_token_inst.client_credentials.return_value = { "access_token": "test_access_token" } ret = get_mgmt_api_token() assert ret == "test_access_token" mock_get_token.assert_called_with( "fake.domain", "test_client_id", client_secret="test_client_secret" ) @pytest.mark.parametrize( "payload, expected, expect_error, description", [ ({}, None, True, "Should fail due to missing fields"), ( { "name": "test-app", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, }, None, True, "Should fail due to missing client_id", ), ( { "name": "test-app", "client_id": "123456", "client_metadata": {"some": "datums"}, }, None, True, "Should fail due to missing client_secret", ), ( { "name": "test-app", "client_id": "123456", "client_secret": "super-secret", }, { "name": "test-app", "client_id": "123456", "client_secret": "super-secret", "client_metadata": {}, }, False, "Should not fail due to missing client_metadata", ), ( { "name": "test-app", "client_id": "123456", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, }, { "name": "test-app", "client_id": "123456", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, }, False, "Should deserialize a valid payload", ), ], ) def test_auth0_client_model( payload: dict[str, Any], expected: Auth0ClientResponse | None, expect_error: bool, description: str, ) -> None: """Test the Auth0ClientResponse schema.""" if expect_error: with pytest.raises(ValidationError): Auth0ClientResponse.model_validate(payload) pytest.fail(description) else: response = Auth0ClientResponse.model_validate(payload) assert response == Auth0ClientResponse.model_validate(expected), description @pytest.mark.parametrize( "payload, expected, expect_error, description", [ ({}, None, True, "Should fail due to missing fields"), ( { "name": "test name", "description": "test description", }, None, True, "Should fail due to missing client_metadata.", ), ( { "description": "test description", "client_metadata": { "m2m_identity_uuid": "423f9f56-ca51-4691-abb9-a839fff00157" }, }, None, True, "Should fail due to missing name.", ), ( { "name": "test name", "client_metadata": { "m2m_identity_uuid": "423f9f56-ca51-4691-abb9-a839fff00157" }, }, None, True, "Should fail due to missing description.", ), ( { "name": "test name", "description": "a" * 1025, "client_metadata": { "m2m_identity_uuid": "423f9f56-ca51-4691-abb9-a839fff00157" }, }, None, True, "Should fail because the description is too long.", ), ( { "name": "test name", "description": "test description", "client_metadata": { "m2m_identity_uuid": "423f9f56-ca51-4691-abb9-a839fff00157" }, }, { "name": "test name", "description": "test description", "client_metadata": { "m2m_identity_uuid": "423f9f56-ca51-4691-abb9-a839fff00157" }, }, False, "Should succeed due to all required fields present.", ), ], ) def test_auth0_client_configuration( payload: dict[str, Any], expected: Auth0ClientResponse | None, expect_error: bool, description: str, ) -> None: """Test the Auth0ClientConfiguration schema.""" if expect_error: with pytest.raises(ValidationError): Auth0ClientConfiguration.model_validate(payload) pytest.fail(description) else: response = Auth0ClientConfiguration.model_validate(payload) assert response == Auth0ClientConfiguration.model_validate( expected ), description @pytest.mark.parametrize( "payload, expected_output, description", [ ( [ { "name": "test-app", "client_id": "123456", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, }, ], { "test-app": Auth0ClientResponse( name="test-app", client_id="123456", client_secret="super-secret", client_metadata={"some": "datums"}, ) }, "Return value should include all entries", ), ( [ { "name": "", "client_id": "123456", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, }, ], {}, "Should skip entries with missing 'name'", ), ], ) def test_clients_dict_by_name( payload: Any, expected_output: dict[str, Auth0ClientResponse], description: str ) -> None: """Test function to index Auth0ClientResponses by client name.""" ret = clients_dict_by_name(payload) assert ret == expected_output, description @patch("m2mconfig.connectors.auth0_connector.time") def test_get_clients( mock_time: MagicMock, mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, ) -> None: """Test the get-clients Auth0 SDK method.""" mock_auth0_mgmt_client_api.all.side_effect = [ { "total": 2, "clients": [ { "name": "test-app-1", "client_id": "123456", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, }, ], }, { "total": 2, "clients": [ { "name": "test-app-2", "client_id": "56789", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, }, ], }, ] auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) auth0_connector.get_clients(page_size=1) assert mock_auth0_mgmt_client_api.all.call_count == 2 mock_auth0_mgmt_client_api.all.assert_has_calls = [ call( fields=GET_CLIENT_FIELDS, page=0, per_page=1, extra_params=GET_CLIENT_EXTRA_PARAMS, ), call( fields=GET_CLIENT_FIELDS, page=1, per_page=1, extra_params=GET_CLIENT_EXTRA_PARAMS, ), ] mock_time.sleep.assert_called_with(config.AUTH_MGMT_RATE_LIMIT_SLEEP_SECONDS) def test_get_client_by_id( mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, ) -> None: """Test get_client_by_id.""" expected_response = { "name": "test-app-1", "client_id": "123456", "client_secret": "super-secret", "client_metadata": {"some": "datums"}, } mock_auth0_mgmt_client_api.get.return_value = expected_response auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) client_response = auth0_connector.get_client_by_id(client_id="123456") assert client_response == Auth0ClientResponse.model_validate(expected_response) mock_auth0_mgmt_client_api.get.assert_called_with( id="123456", fields=GET_CLIENT_FIELDS ) def test_create_client( mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, ) -> None: """Test creating an M2M client.""" expected_create_client_body: dict[str, Any] = Auth0ClientConfiguration( name="pp-m2m-config-test-machine-to-machine", description="Application to generate Machine-to-machine JWTs for: 'pp-m2m-config-test-machine-to-machine'.", client_metadata=ClientMetadata( m2m_identity_uuid=UUID("0faf537c-851c-4d04-bd52-1633016a6f26") ), ).model_dump() expected_create_client_body.update( { "name": "pp-m2m-config-test-machine-to-machine", "client_metadata": { "m2m_identity_uuid": "0faf537c-851c-4d04-bd52-1633016a6f26", "can_impersonate": "false", }, } ) expected_api_response = { "name": "pp-m2m-config-test-machine-to-machine", "client_id": "123456", "client_secret": "super-secret", "client_metadata": { "m2m_identity_uuid": "0faf537c-851c-4d04-bd52-1633016a6f26" }, } mock_auth0_mgmt_client_api.create.return_value = expected_api_response auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) client_response = auth0_connector.create_client( client=RegistryEntry.model_validate( { "name": "pp-m2m-config-test-machine-to-machine", "prod_aws_account": "031099521156", "qa_aws_account": "591204808501", "client_metadata": { "m2m_identity_uuid": "0faf537c-851c-4d04-bd52-1633016a6f26" }, } ) ) assert client_response == Auth0ClientResponse.model_validate(expected_api_response) mock_auth0_mgmt_client_api.create.assert_called_with( body=expected_create_client_body, ) def test_delete_client( mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, ) -> None: mock_auth0_mgmt_client_api.delete.return_value = True auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) delete_response = auth0_connector.delete_client("fake-client-id") mock_auth0_mgmt_client_api.delete.assert_called_with(id="fake-client-id") assert delete_response is True @pytest.mark.parametrize( "mock_get_clients_response, expected_result", [ ({}, None), ( { "pp-m2m-config-test-machine-to-machine": { "name": "pp-m2m-config-test-machine-to-machine", "client_id": "123456", "client_secret": "super-secret", "client_metadata": { "m2m_identity_uuid": "0faf537c-851c-4d04-bd52-1633016a6f26" }, } }, Auth0ClientResponse.model_validate( { "name": "pp-m2m-config-test-machine-to-machine", "client_id": "123456", "client_secret": "super-secret", "client_metadata": { "m2m_identity_uuid": "0faf537c-851c-4d04-bd52-1633016a6f26" }, } ), ), ], ) def test_get_app_by_name( mock_get_clients_response: dict[str, Any], expected_result: Auth0ClientResponse | None, mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, ) -> None: with patch.object( Auth0Connector, "get_clients", return_value=mock_get_clients_response ): auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) response = auth0_connector.get_app_by_name( RegistryEntry.model_validate( { "name": "pp-m2m-config-test-machine-to-machine", "prod_aws_account": "437795906767", "qa_aws_account": "437795906767", "client_metadata": { "m2m_identity_uuid": "1a6f06cd-cc30-4252-8654-cb93a9843b38" }, } ) ) assert response == expected_result @pytest.mark.parametrize( "environment, all_client_grants, expected_m2m_grant", [ pytest.param( config.QA_ENVIRONMENT, [ { "id": "cgr_LaIA1RevCFJ73b91", "client_id": "fake-client-id", "audience": "https://workstation.qaorch.com/api", "scope": [], }, { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://qa-ows.theorchard.io", "scope": [], }, ], { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://qa-ows.theorchard.io", "scope": [], }, id="The grant already exists for the QA audience.", ), pytest.param( config.PROD_ENVIRONMENT, [ { "id": "cgr_LaIA1RevCFJ73b91", "client_id": "fake-client-id", "audience": "https://workstation.theorchard.com/api", "scope": [], }, { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://prod-ows.theorchard.io", "scope": [], }, ], { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://prod-ows.theorchard.io", "scope": [], }, id="The grant already exists for the PROD audience.", ), ], ) def test_create_client_grant__grant_exists( environment: str, all_client_grants: list[dict[str, Any]], expected_m2m_grant: dict[str, Any], mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, monkeypatch: pytest.MonkeyPatch, ) -> None: """Validate create_client_grant does not create a new grant if 1 already exists.""" monkeypatch.setattr(config, "ENVIRONMENT", environment) mock_client_grants_api.all.return_value = ClientGrantsValidator.validate_python( all_client_grants ) auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) grant = auth0_connector.create_client_grant("fake-client-id") assert grant == ClientGrants.model_validate(expected_m2m_grant) mock_client_grants_api.all.assert_called_with(client_id="fake-client-id") mock_client_grants_api.create.assert_not_called() @pytest.mark.parametrize( "environment, expected_audience, created_client_grant_response, expected_m2m_grant", [ pytest.param( config.QA_ENVIRONMENT, constants.QA_AUDIENCE, { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://qa-ows.theorchard.io", "scope": [], }, { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://qa-ows.theorchard.io", "scope": [], }, id="The grant is created for the QA audience.", ), pytest.param( config.PROD_ENVIRONMENT, constants.PROD_AUDIENCE, { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://prod-ows.theorchard.io", "scope": [], }, { "id": "cgr_WghAaLaWK0qqRH0v", "client_id": "fake-client-id", "audience": "https://prod-ows.theorchard.io", "scope": [], }, id="The grant is created for the PROD audience.", ), ], ) def test_create_client_grant__new_grant( environment: str, expected_audience: str, created_client_grant_response: dict[str, Any], expected_m2m_grant: dict[str, Any], mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, monkeypatch: pytest.MonkeyPatch, ) -> None: """Validate create_client_grant can create a new grant when needed.""" monkeypatch.setattr(config, "ENVIRONMENT", environment) mock_client_grants_api.all.return_value = [] mock_client_grants_api.create.return_value = created_client_grant_response auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) grant = auth0_connector.create_client_grant("fake-client-id") assert grant == ClientGrants.model_validate(expected_m2m_grant) mock_client_grants_api.create.assert_called_with( body={"client_id": "fake-client-id", "audience": expected_audience, "scope": []} ) mock_client_grants_api.all.assert_called_with(client_id="fake-client-id") def test_create_client_grant__duplicate_entry_error( mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, monkeypatch: pytest.MonkeyPatch, ) -> None: """Validate that the function raises ClientGrantsCreateError when the auth0 sdk raises an error.""" monkeypatch.setattr(config, "ENVIRONMENT", config.QA_ENVIRONMENT) mock_client_grants_api.all.return_value = [] mock_client_grants_api.create.side_effect = Auth0Error( status_code=409, error_code="testing", message="testing" ) auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) with pytest.raises(ClientGrantsCreateError): auth0_connector.create_client_grant("fake-client-id") pytest.xfail() def test_rotate_client_secret( mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, ) -> None: """Test rotate_client_secret.""" expected_response = { "name": "app-with-old-secret", "client_id": "123456", "client_secret": "rotated-secret", "client_metadata": {"some": "datums"}, } mock_auth0_mgmt_client_api.rotate_secret.return_value = expected_response auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) client_response = auth0_connector.rotate_client_secret("123456") assert client_response == Auth0ClientResponse.model_validate(expected_response) mock_auth0_mgmt_client_api.rotate_secret.assert_called_once_with("123456") def test_update_client_metadata( mock_auth0_mgmt_client_api: MgmtClientsAPI, mock_client_grants_api: ClientGrantsAPI, ) -> None: """Test update_client_metadata.""" expected_response = { "name": "app-with-metadata", "client_id": "123456", "client_secret": "shhh-client-secret-here", "client_metadata": { "m2m_identity_uuid": "1a6f06cd-cc30-4252-8654-cb93a9843b38", "canImpersonate": "true", }, } mock_auth0_mgmt_client_api.update.return_value = expected_response auth0_connector = Auth0Connector( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) client_response = auth0_connector.update_client_metadata( "123456", RegistryEntry.model_validate( { "name": "app-with-metadata", "prod_aws_account": "437795906767", "qa_aws_account": "437795906767", "client_metadata": { "m2m_identity_uuid": UUID("1a6f06cd-cc30-4252-8654-cb93a9843b38"), "additional": "metadata is not updated bc it is not part of schema", "can_impersonate": True, }, } ), ) assert client_response == Auth0ClientResponse.model_validate(expected_response) mock_auth0_mgmt_client_api.update.assert_called_once_with( "123456", body={ "client_metadata": { "m2m_identity_uuid": "1a6f06cd-cc30-4252-8654-cb93a9843b38", "can_impersonate": "true", } }, )