"""Unit tests for sync cli.""" import logging from unittest.mock import MagicMock, call, patch import pytest import typer from mypy_boto3_sts import STSClient from pytest import LogCaptureFixture from m2mconfig.cli.sync import Environment, Sync, all as sync_all, one as sync_one from m2mconfig.connectors.assume_role_secrets_manager import AssumeRoleSecretsManager from m2mconfig.connectors.auth0_connector import Auth0ClientResponse, Auth0Connector from m2mconfig.connectors.client_credentials_manager import ( ClientCredentialsManager, ClientCredentialsSecretScheduledForDeletion, ) from m2mconfig.schemas import NoAwsAccountException, RegistryEntry @pytest.fixture() def mock_auth0_connector() -> MagicMock: """Return mock Auth0Connector.""" return MagicMock(spec=Auth0Connector) @pytest.fixture() def mock_sts_client() -> MagicMock: """Return mock STSClient.""" return MagicMock(spec=STSClient) @pytest.fixture() def mock_arsm() -> MagicMock: """Mock AssumeRoleSecretsManager.""" return MagicMock(spec=AssumeRoleSecretsManager) @pytest.fixture() def mock_ccm() -> MagicMock: """Mock ClientCredentialsManager.""" return MagicMock(spec=ClientCredentialsManager) @patch("m2mconfig.cli.sync.ClientCredentialsManager", autospec=True) @patch("m2mconfig.cli.sync.AssumeRoleSecretsManager", autospec=True) @pytest.mark.parametrize( "secrets_managers,expect_new_arsm", [ pytest.param({}, True, id="empty secrets managers dict"), pytest.param( {"notqanotqanotqa": MagicMock()}, True, id="secrets managers dict doesn't have the aws account", ), pytest.param( {"qaqaqaqaqaqa": MagicMock()}, False, id="secrets managers dict has one for the aws account", ), ], ) def test_sync_get_client_credentials_manager( arsm: MagicMock, ccm: MagicMock, secrets_managers: dict[str, AssumeRoleSecretsManager], expect_new_arsm: bool, mock_qa_aws_account: str, mock_entry_name: str, mock_registry_entry: RegistryEntry, mock_sts_client: MagicMock, mock_arsm: MagicMock, mock_ccm: MagicMock, caplog: LogCaptureFixture, ) -> None: """Test Sync._get_client_credentials_manager.""" arsm.return_value = mock_arsm ccm.return_value = mock_ccm caplog.set_level(logging.INFO) sync = Sync( "qa", MagicMock(), secrets_managers, mock_sts_client, ) result = sync._get_client_credentials_manager(mock_registry_entry) assert result == mock_ccm assert f"[{mock_entry_name}] using AWS Account Id qaqaqaqaqaqa" in caplog.messages if expect_new_arsm: assert ( f"[{mock_entry_name}] adding an AssumeRoleSecretsManager" in caplog.messages ) arsm.assert_called_once_with( "qa", mock_sts_client, mock_qa_aws_account, "assume-auth0-m2m-config", ) ccm.assert_called_once_with( environment="qa", machine_name=mock_entry_name, secrets_manager=mock_arsm, ) assert secrets_managers[mock_qa_aws_account] == mock_arsm else: assert ( f"[{mock_entry_name}] adding an AssumeRoleSecretsManager" not in caplog.messages ) arsm.assert_not_called() ccm.assert_called_once_with( environment="qa", machine_name=mock_entry_name, secrets_manager=secrets_managers[mock_qa_aws_account], ) @pytest.mark.parametrize( "exists, is_valid, overwrite_secret, expected, expected_logs, is_valid_called", [ pytest.param( True, True, True, True, [], True, id="Should be synced because exists, and although valid, overwrite_secret was specified", ), pytest.param( False, False, False, False, ["[best-lambda] secret location does not exist, create it and re-run."], False, id="Should not be synced because secret does not exist", ), pytest.param( False, True, True, False, ["[best-lambda] secret location does not exist, create it and re-run."], False, id="Impossible scenario (exists=False, is_valid=True), in any case, should not be synced because secret does not exist", # noqa: E501 ), pytest.param( True, False, False, True, [], True, id="Should be synced because secret exists but is not valid", ), pytest.param( True, True, False, False, ["[best-lambda] secret is valid and overwrite_secret was not specified."], True, id="Should not be synced because secret is valid but no overwrite_secret", ), ], ) def test_sync_is_client_credentials_manager_syncable( exists: bool, is_valid: bool, overwrite_secret: bool, expected: bool, expected_logs: list[str], is_valid_called: bool, mock_registry_entry: RegistryEntry, mock_ccm: MagicMock, caplog: LogCaptureFixture, ) -> None: """Test Sync._is_client_credentials_manage_syncable.""" mock_ccm.exists.return_value = exists mock_ccm.is_valid.return_value = is_valid caplog.set_level(logging.INFO) sync = Sync( "qa", MagicMock(), {}, MagicMock(), ) actual = sync._is_client_credentials_manager_syncable( mock_registry_entry, mock_ccm, overwrite_secret=overwrite_secret, ) assert actual == expected assert expected_logs == caplog.messages mock_ccm.exists.assert_called_once() if is_valid_called: mock_ccm.is_valid.assert_called_once() else: mock_ccm.is_valid.assert_not_called() def mock_auth0_client_response() -> Auth0ClientResponse: """Mock Auth0ClientResponse.""" return Auth0ClientResponse( name="test-app", client_id="123456", client_secret="super-secret", client_metadata={"some": "datums"}, ) def mock_auth0_rotated_client_response() -> Auth0ClientResponse: """Mock Auth0ClientResponse after secret rotation.""" return Auth0ClientResponse( name="test-app", client_id="123456", client_secret="rotated-super-secret", client_metadata={"some": "datums"}, ) @pytest.mark.parametrize( "get_app_by_name, dry_run, expected, expect_new_client, expected_logs", [ pytest.param( None, True, None, False, ["[best-lambda] creating auth0 client"], id="client does not exist, but dry run means don't create it", ), pytest.param( None, False, mock_auth0_client_response(), True, [ "[best-lambda] creating auth0 client", "[best-lambda] auth0 client created", ], id="client does not exist, so create it", ), pytest.param( mock_auth0_client_response(), False, mock_auth0_client_response(), False, ["[best-lambda] auth0 client already exists"], id="client exists, no need to create it", ), pytest.param( mock_auth0_client_response(), True, mock_auth0_client_response(), False, ["[best-lambda] auth0 client already exists"], id="client exists, dry run is irrelevant", ), ], ) def test_sync_get_or_create_auth0_client( get_app_by_name: Auth0ClientResponse | None, dry_run: bool, expected: Auth0ClientResponse | None, expect_new_client: bool, expected_logs: list[str], mock_registry_entry: RegistryEntry, mock_auth0_connector: MagicMock, caplog: LogCaptureFixture, ) -> None: """Test Sync._get_or_create_auth0_client.""" caplog.set_level(logging.INFO) mock_auth0_connector.get_app_by_name.return_value = get_app_by_name mock_auth0_connector.create_client.return_value = expected sync = Sync( "qa", mock_auth0_connector, {}, MagicMock(), ) actual = sync._get_or_create_auth0_client( mock_registry_entry, dry_run=dry_run, ) assert actual == expected mock_auth0_connector.get_app_by_name.assert_called_once_with(mock_registry_entry) assert expected_logs == caplog.messages if expect_new_client: mock_auth0_connector.create_client.assert_called_once_with(mock_registry_entry) else: mock_auth0_connector.create_client.assert_not_called() if not dry_run: mock_auth0_connector.create_client_grant.assert_called_with(client_id="123456") mock_auth0_connector.update_client_metadata.assert_called_with( "123456", mock_registry_entry ) else: mock_auth0_connector.create_client_grant.assert_not_called() mock_auth0_connector.update_client_metadata.assert_not_called() @pytest.mark.parametrize( "get_app_by_name, dry_run, expected, expect_rotation, expected_logs", [ pytest.param( None, True, None, False, ["[best-lambda] auth0 client does not exist, nothing to rotate"], id="client does not exist so there is nothing to rotate", ), pytest.param( mock_auth0_client_response(), True, mock_auth0_client_response(), False, [], id="dry run - secret is not rotated", ), pytest.param( mock_auth0_client_response(), False, mock_auth0_client_response(), True, [ "[best-lambda] rotating secret for auth0 client", "[best-lambda] secret is rotated", ], id="client exists, rotate secret", ), ], ) def test_sync_rotate_client_secret( get_app_by_name: Auth0ClientResponse | None, dry_run: bool, expected: Auth0ClientResponse | None, expect_rotation: bool, expected_logs: list[str], mock_registry_entry: RegistryEntry, mock_auth0_connector: MagicMock, caplog: LogCaptureFixture, ) -> None: """Test Sync._rotate_client_secret.""" caplog.set_level(logging.INFO) mock_auth0_connector.get_app_by_name.return_value = get_app_by_name mock_auth0_connector.rotate_client_secret.return_value = expected sync = Sync( "qa", mock_auth0_connector, {}, MagicMock(), ) actual = sync._rotate_client_secret( mock_registry_entry, dry_run=dry_run, ) assert actual == expected mock_auth0_connector.get_app_by_name.assert_called_once_with(mock_registry_entry) assert expected_logs == caplog.messages if expect_rotation: assert get_app_by_name mock_auth0_connector.rotate_client_secret.assert_called_once_with( client_id=get_app_by_name.client_id ) else: mock_auth0_connector.rotate_client_secret.assert_not_called() @patch.object(Sync, "_rotate_client_secret") @patch.object(Sync, "_get_or_create_auth0_client") @patch.object(Sync, "_is_client_credentials_manager_syncable") @patch.object(Sync, "_get_client_credentials_manager") @pytest.mark.parametrize( "is_syncable, is_auth0_client_get_create_called", [ pytest.param( True, True, id="Secret is syncable, _get_or_create_auth0_client should be called.", ), pytest.param(False, False, id="Secret is not syncable, stop"), ], ) @pytest.mark.parametrize( "mock_auth0_client_response, dry_run, expect_set", [ pytest.param(None, True, False, id="no auth0 client, nothing to set"), pytest.param( None, False, False, id="no auth0 client, nothing to set, regardless of dry run", ), pytest.param( mock_auth0_client_response(), False, True, id="auth0 client returned and it is not a dry run", ), pytest.param( mock_auth0_client_response(), True, False, id="auth0 client returned but it is a dry run", ), ], ) def test_sync_registry_entry( mock__get_client_credentials_manager: MagicMock, mock__is_client_credentials_manager_syncable: MagicMock, mock__get_or_create_auth0_client: MagicMock, mock__rotate_client_secret: MagicMock, is_syncable: bool, is_auth0_client_get_create_called: bool, mock_auth0_client_response: Auth0ClientResponse, dry_run: bool, expect_set: bool, mock_registry_entry: RegistryEntry, mock_auth0_connector: MagicMock, mock_sts_client: MagicMock, mock_ccm: MagicMock, caplog: LogCaptureFixture, ) -> None: """Test sync_registry_entry creates and uses an AssumeRoleSecretsManager.""" caplog.set_level(logging.INFO) mock__get_client_credentials_manager.return_value = mock_ccm mock__is_client_credentials_manager_syncable.return_value = is_syncable mock__get_or_create_auth0_client.return_value = mock_auth0_client_response sync = Sync( "qa", mock_auth0_connector, {}, mock_sts_client, ) sync.sync_registry_entry( mock_registry_entry, overwrite_secret=True, rotate_secret=False, dry_run=dry_run, ) mock__get_client_credentials_manager.assert_called_once_with(mock_registry_entry) mock__is_client_credentials_manager_syncable.assert_called_once_with( mock_registry_entry, mock_ccm, overwrite_secret=True, ) mock__rotate_client_secret.assert_not_called() if is_auth0_client_get_create_called: mock__get_or_create_auth0_client.assert_called_once_with( mock_registry_entry, dry_run=dry_run, ) if not mock_auth0_client_response: assert "[best-lambda] no auth0 client created or found" in caplog.messages if expect_set: assert ( "[best-lambda] setting auth0 client credentials in SecretsManager" in caplog.messages ) mock_ccm.set.assert_called_once_with( "123456", "super-secret", ) else: mock_ccm.set.assert_not_called() else: mock__get_or_create_auth0_client.assert_not_called() mock_ccm.set.assert_not_called() @patch.object(Sync, "_rotate_client_secret") @patch.object(Sync, "_get_or_create_auth0_client") @patch.object(Sync, "_is_client_credentials_manager_syncable") @patch.object(Sync, "_get_client_credentials_manager") @pytest.mark.parametrize( "overwrite_secret", [ pytest.param(True, id="overwrite_secret flag stays set to True"), pytest.param(False, id="overwrite_secret flag gets set so False"), ], ) @pytest.mark.parametrize( "mock_auth0_client_response, mock_auth0_rotated_client_response, dry_run, expect_set", [ pytest.param( None, None, True, False, id="no auth0 client, nothing to rotate, nothing to set", ), pytest.param( None, None, False, False, id="no auth0 client, nothing to rotate, regardless of dry run", ), pytest.param( mock_auth0_client_response(), mock_auth0_rotated_client_response(), False, True, id="auth0 client returned, rotated and it is not a dry run", ), pytest.param( mock_auth0_client_response(), None, False, False, id="auth0 client returned, rotation returns none, so nothing to set", ), pytest.param( mock_auth0_client_response(), None, True, False, id=""" auth0 client returned, rotate fn takes dry flag so would return None, but nothing is done because it is a dry run""", ), ], ) def test_sync_registry_entry_rotate_secret( mock__get_client_credentials_manager: MagicMock, mock__is_client_credentials_manager_syncable: MagicMock, mock__get_or_create_auth0_client: MagicMock, mock__rotate_client_secret: MagicMock, overwrite_secret: bool, mock_auth0_client_response: Auth0ClientResponse, mock_auth0_rotated_client_response: Auth0ClientResponse, dry_run: bool, expect_set: bool, mock_registry_entry: RegistryEntry, mock_auth0_connector: MagicMock, mock_sts_client: MagicMock, mock_ccm: MagicMock, caplog: LogCaptureFixture, ) -> None: """Test sync_registry_entry uses rotate_secret.""" caplog.set_level(logging.INFO) mock__get_client_credentials_manager.return_value = mock_ccm mock__is_client_credentials_manager_syncable.return_value = True mock__get_or_create_auth0_client.return_value = mock_auth0_client_response mock__rotate_client_secret.return_value = mock_auth0_rotated_client_response sync = Sync( "qa", mock_auth0_connector, {}, mock_sts_client, ) sync.sync_registry_entry( mock_registry_entry, overwrite_secret=overwrite_secret, rotate_secret=True, dry_run=dry_run, ) mock__get_client_credentials_manager.assert_called_once_with(mock_registry_entry) mock__is_client_credentials_manager_syncable.assert_called_once_with( mock_registry_entry, mock_ccm, overwrite_secret=True, ) mock__get_or_create_auth0_client.assert_called_once_with( mock_registry_entry, dry_run=dry_run, ) if mock_auth0_client_response: mock__rotate_client_secret.assert_called_once_with( mock_registry_entry, dry_run=dry_run ) else: mock__rotate_client_secret.assert_not_called() if not mock_auth0_client_response: assert "[best-lambda] no auth0 client created or found" in caplog.messages else: assert "[best-lambda] no auth0 client created or found" not in caplog.messages if not mock_auth0_rotated_client_response: assert ( "[best-lambda] no auth0 client returned with secret rotation call" in caplog.messages ) else: assert ( "[best-lambda] no auth0 client returned with secret rotation call" not in caplog.messages ) if expect_set: assert ( "[best-lambda] setting auth0 client credentials in SecretsManager" in caplog.messages ) mock_ccm.set.assert_called_once_with( "123456", "rotated-super-secret", ) else: mock_ccm.set.assert_not_called() @patch("m2mconfig.cli.sync.Auth0Connector", autospec=True) @patch("m2mconfig.cli.sync.get_auth0_mgmt_client_api") @patch("m2mconfig.cli.sync.get_mgmt_api_token") @patch("m2mconfig.cli.sync.get_client_grants_api") @patch("m2mconfig.cli.sync.validate_config") @patch("m2mconfig.cli.sync.boto3") @patch("m2mconfig.cli.sync.Sync", autospec=True) @pytest.mark.parametrize("overwrite_secret", [(True), (False)]) @pytest.mark.parametrize("rotate_secret", [(True), (False)]) @pytest.mark.parametrize("dry_run", [(True), (False)]) @pytest.mark.parametrize( "environment, env_str", [(Environment.qa, "qa"), (Environment.prod, "prod"), (Environment.uat, "uat")], ) def test_cli_sync_all( sync: MagicMock, mock_boto3: MagicMock, mock_validate_config: MagicMock, mock_get_client_grants_api: MagicMock, mock_get_mgmt_api_token: MagicMock, mock_get_auth0_mgmt_client_api: MagicMock, auth0c: MagicMock, environment: Environment, env_str: str, dry_run: bool, rotate_secret: bool, overwrite_secret: bool, mock_sts_client: MagicMock, mock_auth0_connector: MagicMock, mock_registry_entry: RegistryEntry, caplog: pytest.LogCaptureFixture, ) -> None: """Test cli command to sync all.""" caplog.set_level(logging.INFO) no_aws_account_registry_entry = MagicMock(spec=RegistryEntry) no_aws_account_registry_entry.name = "this-has-no-aws-account" client_creds_being_deleted_registry_entry = MagicMock(spec=RegistryEntry) client_creds_being_deleted_registry_entry.name = ( "client-credentials-secret-scheduled-for-deletion" ) fourth_registry_entry = MagicMock(spec=RegistryEntry) exception_registry_entry = MagicMock(spec=RegistryEntry) never_touched_registry_entry = MagicMock(spec=RegistryEntry) mock_validate_config.return_value = [ mock_registry_entry, no_aws_account_registry_entry, client_creds_being_deleted_registry_entry, fourth_registry_entry, exception_registry_entry, never_touched_registry_entry, ] mock_auth0_mgmt_client_api = MagicMock() mock_get_auth0_mgmt_client_api.return_value = mock_auth0_mgmt_client_api mock_get_mgmt_api_token.return_value = "test.mgmt.token" mock_client_grants_api = MagicMock() mock_get_client_grants_api.return_value = mock_client_grants_api auth0c.return_value = mock_auth0_connector mock_boto3.client.return_value = mock_sts_client mock_sync = MagicMock(spec=Sync) sync.return_value = mock_sync mock_sync.sync_registry_entry = MagicMock( side_effect=[ (None), # mock_registry_entry call (NoAwsAccountException()), # no_aws_account_registry_entry call ( ClientCredentialsSecretScheduledForDeletion() ), # client_creds_being_deleted_registry_entry call (None), # fourth_registry_entry call (ValueError("unhandled")), ] ) with pytest.raises(typer.Exit): sync_all( environment, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, dry_run=dry_run, ) mock_validate_config.assert_called_once_with("./m2m.json") mock_get_auth0_mgmt_client_api.assert_called_once() auth0c.assert_called_once_with( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) mock_boto3.client.assert_called_once_with("sts") sync.assert_called_once_with( env_str, mock_auth0_connector, {}, mock_sts_client, ) assert mock_sync.sync_registry_entry.call_count == 5 mock_sync.sync_registry_entry.assert_has_calls( [ call( mock_registry_entry, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, dry_run=dry_run, ), call( no_aws_account_registry_entry, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, dry_run=dry_run, ), call( client_creds_being_deleted_registry_entry, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, dry_run=dry_run, ), call( fourth_registry_entry, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, dry_run=dry_run, ), call( exception_registry_entry, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, dry_run=dry_run, ), ] ) assert ( caplog.records[0].message == f"[this-has-no-aws-account] No AWS Account configured for {env_str} environment" ) assert ( caplog.records[1].message == f"[{client_creds_being_deleted_registry_entry.name}] secret is scheduled for deletion, skipping" ) assert caplog.records[2].message == "Unhandled error in 'all': unhandled" @patch("m2mconfig.cli.sync.Auth0Connector", autospec=True) @patch("m2mconfig.cli.sync.get_auth0_mgmt_client_api") @patch("m2mconfig.cli.sync.get_mgmt_api_token") @patch("m2mconfig.cli.sync.get_client_grants_api") @patch("m2mconfig.cli.sync.validate_config") @patch("m2mconfig.cli.sync.boto3") @patch("m2mconfig.cli.sync.Sync", autospec=True) @pytest.mark.parametrize("rotate_secret", [(True), (False)]) @pytest.mark.parametrize("overwrite_secret", [(True), (False)]) @pytest.mark.parametrize( "environment, env_str", [(Environment.qa, "qa"), (Environment.prod, "prod"), (Environment.uat, "uat")], ) @pytest.mark.parametrize( "machine_name, expect_calls", [("best-lambda", True), ("no-lambda", False)] ) def test_cli_sync_one( sync: MagicMock, mock_boto3: MagicMock, mock_validate_config: MagicMock, mock_get_client_grants_api: MagicMock, mock_get_mgmt_api_token: MagicMock, mock_get_auth0_mgmt_client_api: MagicMock, auth0c: MagicMock, machine_name: str, expect_calls: bool, environment: Environment, env_str: str, rotate_secret: bool, overwrite_secret: bool, mock_sts_client: MagicMock, mock_auth0_connector: MagicMock, mock_registry_entry: RegistryEntry, ) -> None: """Test cli command to sync one.""" mock_validate_config.return_value = [ mock_registry_entry, ] mock_auth0_mgmt_client_api = MagicMock() mock_get_auth0_mgmt_client_api.return_value = mock_auth0_mgmt_client_api mock_get_mgmt_api_token.return_value = "test.mgmt.token" mock_client_grants_api = MagicMock() mock_get_client_grants_api.return_value = mock_client_grants_api auth0c.return_value = mock_auth0_connector mock_boto3.client.return_value = mock_sts_client mock_sync = MagicMock(spec=Sync) sync.return_value = mock_sync sync_one( environment, machine_name, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, ) mock_validate_config.assert_called_once_with("./m2m.json") if expect_calls: mock_get_auth0_mgmt_client_api.assert_called_once() auth0c.assert_called_once_with( auth0_mgmt_client_api=mock_auth0_mgmt_client_api, client_grants_api=mock_client_grants_api, ) mock_boto3.client.assert_called_once_with("sts") sync.assert_called_once_with( env_str, mock_auth0_connector, {}, mock_sts_client, ) mock_sync.sync_registry_entry.assert_called_once_with( mock_registry_entry, overwrite_secret=overwrite_secret, rotate_secret=rotate_secret, ) else: mock_get_auth0_mgmt_client_api.assert_not_called() auth0c.assert_not_called() mock_boto3.client.assert_not_called() sync.assert_not_called() mock_sync.sync_registry_entry.assert_not_called()