From 6b861135346910de32cb429e60546ad2a4e79179 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Sun, 11 Jun 2023 14:45:26 +0000 Subject: [PATCH] Feature: AppConfig (#6391) --- IMPLEMENTATION_COVERAGE.md | 50 +++- docs/docs/services/appconfig.rst | 77 +++++ moto/__init__.py | 1 + moto/appconfig/__init__.py | 5 + moto/appconfig/exceptions.py | 19 ++ moto/appconfig/models.py | 281 ++++++++++++++++++ moto/appconfig/responses.py | 169 +++++++++++ moto/appconfig/urls.py | 22 ++ moto/backend_index.py | 1 + .../terraform-tests.success.txt | 6 + tests/test_appconfig/__init__.py | 0 .../test_appconfig_applications.py | 83 ++++++ .../test_appconfig_config_profiles.py | 134 +++++++++ .../test_appconfig_hosted_config_versions.py | 88 ++++++ 14 files changed, 935 insertions(+), 1 deletion(-) create mode 100644 docs/docs/services/appconfig.rst create mode 100644 moto/appconfig/__init__.py create mode 100644 moto/appconfig/exceptions.py create mode 100644 moto/appconfig/models.py create mode 100644 moto/appconfig/responses.py create mode 100644 moto/appconfig/urls.py create mode 100644 tests/test_appconfig/__init__.py create mode 100644 tests/test_appconfig/test_appconfig_applications.py create mode 100644 tests/test_appconfig/test_appconfig_config_profiles.py create mode 100644 tests/test_appconfig/test_appconfig_hosted_config_versions.py diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md index 693cbd695..9c42d8337 100644 --- a/IMPLEMENTATION_COVERAGE.md +++ b/IMPLEMENTATION_COVERAGE.md @@ -280,6 +280,55 @@ - [X] update_vpc_link +## appconfig +
+34% implemented + +- [X] create_application +- [X] create_configuration_profile +- [ ] create_deployment_strategy +- [ ] create_environment +- [ ] create_extension +- [ ] create_extension_association +- [X] create_hosted_configuration_version +- [X] delete_application +- [X] delete_configuration_profile +- [ ] delete_deployment_strategy +- [ ] delete_environment +- [ ] delete_extension +- [ ] delete_extension_association +- [X] delete_hosted_configuration_version +- [X] get_application +- [ ] get_configuration +- [X] get_configuration_profile +- [ ] get_deployment +- [ ] get_deployment_strategy +- [ ] get_environment +- [ ] get_extension +- [ ] get_extension_association +- [X] get_hosted_configuration_version +- [ ] list_applications +- [X] list_configuration_profiles +- [ ] list_deployment_strategies +- [ ] list_deployments +- [ ] list_environments +- [ ] list_extension_associations +- [ ] list_extensions +- [ ] list_hosted_configuration_versions +- [X] list_tags_for_resource +- [ ] start_deployment +- [ ] stop_deployment +- [X] tag_resource +- [X] untag_resource +- [X] update_application +- [X] update_configuration_profile +- [ ] update_deployment_strategy +- [ ] update_environment +- [ ] update_extension +- [ ] update_extension_association +- [ ] validate_configuration +
+ ## application-autoscaling
69% implemented @@ -7121,7 +7170,6 @@ - amplifybackend - amplifyuibuilder - apigatewaymanagementapi -- appconfig - appconfigdata - appflow - appintegrations diff --git a/docs/docs/services/appconfig.rst b/docs/docs/services/appconfig.rst new file mode 100644 index 000000000..c06901dba --- /dev/null +++ b/docs/docs/services/appconfig.rst @@ -0,0 +1,77 @@ +.. _implementedservice_appconfig: + +.. |start-h3| raw:: html + +

+ +.. |end-h3| raw:: html + +

+ +========= +appconfig +========= + +.. autoclass:: moto.appconfig.models.AppConfigBackend + +|start-h3| Example usage |end-h3| + +.. sourcecode:: python + + @mock_appconfig + def test_appconfig_behaviour: + boto3.client("appconfig") + ... + + + +|start-h3| Implemented features for this service |end-h3| + +- [X] create_application +- [X] create_configuration_profile +- [ ] create_deployment_strategy +- [ ] create_environment +- [ ] create_extension +- [ ] create_extension_association +- [X] create_hosted_configuration_version + + The LatestVersionNumber-parameter is not yet implemented + + +- [X] delete_application +- [X] delete_configuration_profile +- [ ] delete_deployment_strategy +- [ ] delete_environment +- [ ] delete_extension +- [ ] delete_extension_association +- [X] delete_hosted_configuration_version +- [X] get_application +- [ ] get_configuration +- [X] get_configuration_profile +- [ ] get_deployment +- [ ] get_deployment_strategy +- [ ] get_environment +- [ ] get_extension +- [ ] get_extension_association +- [X] get_hosted_configuration_version +- [ ] list_applications +- [X] list_configuration_profiles +- [ ] list_deployment_strategies +- [ ] list_deployments +- [ ] list_environments +- [ ] list_extension_associations +- [ ] list_extensions +- [ ] list_hosted_configuration_versions +- [X] list_tags_for_resource +- [ ] start_deployment +- [ ] stop_deployment +- [X] tag_resource +- [X] untag_resource +- [X] update_application +- [X] update_configuration_profile +- [ ] update_deployment_strategy +- [ ] update_environment +- [ ] update_extension +- [ ] update_extension_association +- [ ] validate_configuration + diff --git a/moto/__init__.py b/moto/__init__.py index 93c168728..f1baa7596 100644 --- a/moto/__init__.py +++ b/moto/__init__.py @@ -20,6 +20,7 @@ mock_acmpca = lazy_load(".acmpca", "mock_acmpca", boto3_name="acm-pca") mock_amp = lazy_load(".amp", "mock_amp") mock_apigateway = lazy_load(".apigateway", "mock_apigateway") mock_apigatewayv2 = lazy_load(".apigatewayv2", "mock_apigatewayv2") +mock_appconfig = lazy_load(".appconfig", "mock_appconfig") mock_appsync = lazy_load(".appsync", "mock_appsync") mock_athena = lazy_load(".athena", "mock_athena") mock_applicationautoscaling = lazy_load( diff --git a/moto/appconfig/__init__.py b/moto/appconfig/__init__.py new file mode 100644 index 000000000..50f09c2f6 --- /dev/null +++ b/moto/appconfig/__init__.py @@ -0,0 +1,5 @@ +"""appconfig module initialization; sets value for base decorator.""" +from .models import appconfig_backends +from ..core.models import base_decorator + +mock_appconfig = base_decorator(appconfig_backends) diff --git a/moto/appconfig/exceptions.py b/moto/appconfig/exceptions.py new file mode 100644 index 000000000..ec0a6967b --- /dev/null +++ b/moto/appconfig/exceptions.py @@ -0,0 +1,19 @@ +"""Exceptions raised by the appconfig service.""" +from moto.core.exceptions import JsonRESTError + + +class AppNotFoundException(JsonRESTError): + def __init__(self) -> None: + super().__init__("ResourceNotFoundException", "Application not found") + + +class ConfigurationProfileNotFound(JsonRESTError): + def __init__(self) -> None: + super().__init__("ResourceNotFoundException", "ConfigurationProfile not found") + + +class ConfigurationVersionNotFound(JsonRESTError): + def __init__(self) -> None: + super().__init__( + "ResourceNotFoundException", "HostedConfigurationVersion not found" + ) diff --git a/moto/appconfig/models.py b/moto/appconfig/models.py new file mode 100644 index 000000000..e993b2045 --- /dev/null +++ b/moto/appconfig/models.py @@ -0,0 +1,281 @@ +from moto.core import BaseBackend, BackendDict, BaseModel +from moto.moto_api._internal import mock_random +from moto.utilities.tagging_service import TaggingService +from typing import Any, Dict, List, Iterable, Optional +from .exceptions import ( + AppNotFoundException, + ConfigurationProfileNotFound, + ConfigurationVersionNotFound, +) + + +class HostedConfigurationVersion(BaseModel): + def __init__( + self, + app_id: str, + config_id: str, + version: int, + description: str, + content: str, + content_type: str, + version_label: str, + ): + self.app_id = app_id + self.config_id = config_id + self.version = version + self.description = description + self.content = content + self.content_type = content_type + self.version_label = version_label + + def get_headers(self) -> Dict[str, Any]: + return { + "application-id": self.app_id, + "configuration-profile-id": self.config_id, + "version-number": self.version, + "description": self.description, + "content-type": self.content_type, + "VersionLabel": self.version_label, + } + + +class ConfigurationProfile(BaseModel): + def __init__( + self, + application_id: str, + name: str, + region: str, + account_id: str, + description: str, + location_uri: str, + retrieval_role_arn: str, + validators: List[Dict[str, str]], + _type: str, + ): + self.id = mock_random.get_random_hex(7) + self.arn = f"arn:aws:appconfig:{region}:{account_id}:application/{application_id}/configurationprofile/{self.id}" + self.application_id = application_id + self.name = name + self.description = description + self.location_uri = location_uri + self.retrieval_role_arn = retrieval_role_arn + self.validators = validators + self._type = _type + self.config_versions: Dict[int, HostedConfigurationVersion] = dict() + + def create_version( + self, + app_id: str, + config_id: str, + description: str, + content: str, + content_type: str, + version_label: str, + ) -> HostedConfigurationVersion: + if self.config_versions: + version = sorted(self.config_versions.keys())[-1] + 1 + else: + version = 1 + self.config_versions[version] = HostedConfigurationVersion( + app_id=app_id, + config_id=config_id, + version=version, + description=description, + content=content, + content_type=content_type, + version_label=version_label, + ) + return self.config_versions[version] + + def get_version(self, version: int) -> HostedConfigurationVersion: + if version not in self.config_versions: + raise ConfigurationVersionNotFound + return self.config_versions[version] + + def delete_version(self, version: int) -> None: + self.config_versions.pop(version) + + def to_json(self) -> Dict[str, Any]: + return { + "Id": self.id, + "Name": self.name, + "ApplicationId": self.application_id, + "Description": self.description, + "LocationUri": self.location_uri, + "RetrievalRoleArn": self.retrieval_role_arn, + "Validators": self.validators, + "Type": self._type, + } + + +class Application(BaseModel): + def __init__( + self, name: str, description: Optional[str], region: str, account_id: str + ): + self.id = mock_random.get_random_hex(7) + self.arn = f"arn:aws:appconfig:{region}:{account_id}:application/{self.id}" + self.name = name + self.description = description + + self.config_profiles: Dict[str, ConfigurationProfile] = dict() + + def to_json(self) -> Dict[str, Any]: + return { + "Id": self.id, + "Name": self.name, + "Description": self.description, + } + + +class AppConfigBackend(BaseBackend): + """Implementation of AppConfig APIs.""" + + def __init__(self, region_name: str, account_id: str): + super().__init__(region_name, account_id) + self.applications: Dict[str, Application] = dict() + self.tagger = TaggingService() + + def create_application( + self, name: str, description: Optional[str], tags: Dict[str, str] + ) -> Application: + app = Application( + name, description, region=self.region_name, account_id=self.account_id + ) + self.applications[app.id] = app + self.tag_resource(app.arn, tags) + return app + + def delete_application(self, app_id: str) -> None: + self.applications.pop(app_id, None) + + def get_application(self, app_id: str) -> Application: + if app_id not in self.applications: + raise AppNotFoundException + return self.applications[app_id] + + def update_application( + self, application_id: str, name: str, description: str + ) -> Application: + app = self.get_application(application_id) + if name is not None: + app.name = name + if description is not None: + app.description = description + return app + + def create_configuration_profile( + self, + application_id: str, + name: str, + description: str, + location_uri: str, + retrieval_role_arn: str, + validators: List[Dict[str, str]], + _type: str, + tags: Dict[str, str], + ) -> ConfigurationProfile: + config_profile = ConfigurationProfile( + application_id=application_id, + name=name, + region=self.region_name, + account_id=self.account_id, + description=description, + location_uri=location_uri, + retrieval_role_arn=retrieval_role_arn, + validators=validators, + _type=_type, + ) + self.tag_resource(config_profile.arn, tags) + self.get_application(application_id).config_profiles[ + config_profile.id + ] = config_profile + return config_profile + + def delete_configuration_profile(self, app_id: str, config_profile_id: str) -> None: + self.get_application(app_id).config_profiles.pop(config_profile_id) + + def get_configuration_profile( + self, app_id: str, config_profile_id: str + ) -> ConfigurationProfile: + app = self.get_application(app_id) + if config_profile_id not in app.config_profiles: + raise ConfigurationProfileNotFound + return app.config_profiles[config_profile_id] + + def update_configuration_profile( + self, + application_id: str, + config_profile_id: str, + name: str, + description: str, + retrieval_role_arn: str, + validators: List[Dict[str, str]], + ) -> ConfigurationProfile: + config_profile = self.get_configuration_profile( + application_id, config_profile_id + ) + if name is not None: + config_profile.name = name + if description is not None: + config_profile.description = description + if retrieval_role_arn is not None: + config_profile.retrieval_role_arn = retrieval_role_arn + if validators is not None: + config_profile.validators = validators + return config_profile + + def list_configuration_profiles( + self, app_id: str + ) -> Iterable[ConfigurationProfile]: + app = self.get_application(app_id) + return app.config_profiles.values() + + def create_hosted_configuration_version( + self, + app_id: str, + config_profile_id: str, + description: str, + content: str, + content_type: str, + version_label: str, + ) -> HostedConfigurationVersion: + """ + The LatestVersionNumber-parameter is not yet implemented + """ + profile = self.get_configuration_profile(app_id, config_profile_id) + return profile.create_version( + app_id=app_id, + config_id=config_profile_id, + description=description, + content=content, + content_type=content_type, + version_label=version_label, + ) + + def get_hosted_configuration_version( + self, app_id: str, config_profile_id: str, version: int + ) -> HostedConfigurationVersion: + profile = self.get_configuration_profile( + app_id=app_id, config_profile_id=config_profile_id + ) + return profile.get_version(version) + + def delete_hosted_configuration_version( + self, app_id: str, config_profile_id: str, version: int + ) -> None: + profile = self.get_configuration_profile( + app_id=app_id, config_profile_id=config_profile_id + ) + profile.delete_version(version=version) + + def list_tags_for_resource(self, arn: str) -> Dict[str, str]: + return self.tagger.get_tag_dict_for_resource(arn) + + def tag_resource(self, arn: str, tags: Dict[str, str]) -> None: + self.tagger.tag_resource(arn, TaggingService.convert_dict_to_tags_input(tags)) + + def untag_resource(self, arn: str, tag_keys: List[str]) -> None: + self.tagger.untag_resource_using_names(arn, tag_keys) + + +appconfig_backends = BackendDict(AppConfigBackend, "appconfig") diff --git a/moto/appconfig/responses.py b/moto/appconfig/responses.py new file mode 100644 index 000000000..78bff4e1f --- /dev/null +++ b/moto/appconfig/responses.py @@ -0,0 +1,169 @@ +import json + +from moto.core.responses import BaseResponse +from .models import appconfig_backends, AppConfigBackend +from typing import Any, Dict, Tuple +from urllib.parse import unquote + + +class AppConfigResponse(BaseResponse): + def tags(self, request: Any, full_url: str, headers: Any) -> str: # type: ignore[return] + self.setup_class(request, full_url, headers) + if request.method == "GET": + return self.list_tags_for_resource() + if request.method == "POST": + return self.tag_resource() + if request.method == "DELETE": + return self.untag_resource() + + def __init__(self) -> None: + super().__init__(service_name="appconfig") + + @property + def appconfig_backend(self) -> AppConfigBackend: + return appconfig_backends[self.current_account][self.region] + + def create_application(self) -> str: + name = self._get_param("Name") + description = self._get_param("Description") + tags = self._get_param("Tags") + app = self.appconfig_backend.create_application( + name=name, + description=description, + tags=tags, + ) + return json.dumps(app.to_json()) + + def delete_application(self) -> str: + app_id = self._get_param("ApplicationId") + self.appconfig_backend.delete_application(app_id) + return "{}" + + def get_application(self) -> str: + app_id = self._get_param("ApplicationId") + app = self.appconfig_backend.get_application(app_id) + return json.dumps(app.to_json()) + + def update_application(self) -> str: + app_id = self._get_param("ApplicationId") + name = self._get_param("Name") + description = self._get_param("Description") + app = self.appconfig_backend.update_application( + application_id=app_id, + name=name, + description=description, + ) + return json.dumps(app.to_json()) + + def create_configuration_profile(self) -> str: + app_id = self._get_param("ApplicationId") + name = self._get_param("Name") + description = self._get_param("Description") + location_uri = self._get_param("LocationUri") + retrieval_role_arn = self._get_param("RetrievalRoleArn") + validators = self._get_param("Validators") + _type = self._get_param("Type") + tags = self._get_param("Tags") + config_profile = self.appconfig_backend.create_configuration_profile( + application_id=app_id, + name=name, + description=description, + location_uri=location_uri, + retrieval_role_arn=retrieval_role_arn, + validators=validators, + _type=_type, + tags=tags, + ) + return json.dumps(config_profile.to_json()) + + def delete_configuration_profile(self) -> str: + app_id = self._get_param("ApplicationId") + config_profile_id = self._get_param("ConfigurationProfileId") + self.appconfig_backend.delete_configuration_profile(app_id, config_profile_id) + return "{}" + + def get_configuration_profile(self) -> str: + app_id = self._get_param("ApplicationId") + config_profile_id = self._get_param("ConfigurationProfileId") + config_profile = self.appconfig_backend.get_configuration_profile( + app_id, config_profile_id + ) + return json.dumps(config_profile.to_json()) + + def update_configuration_profile(self) -> str: + app_id = self._get_param("ApplicationId") + config_profile_id = self._get_param("ConfigurationProfileId") + name = self._get_param("Name") + description = self._get_param("Description") + retrieval_role_arn = self._get_param("RetrievalRoleArn") + validators = self._get_param("Validators") + config_profile = self.appconfig_backend.update_configuration_profile( + application_id=app_id, + config_profile_id=config_profile_id, + name=name, + description=description, + retrieval_role_arn=retrieval_role_arn, + validators=validators, + ) + return json.dumps(config_profile.to_json()) + + def list_configuration_profiles(self) -> str: + app_id = self._get_param("ApplicationId") + profiles = self.appconfig_backend.list_configuration_profiles(app_id) + return json.dumps({"Items": [p.to_json() for p in profiles]}) + + def list_tags_for_resource(self) -> str: + arn = unquote(self.path.split("/tags/")[-1]) + tags = self.appconfig_backend.list_tags_for_resource(arn) + return json.dumps({"Tags": tags}) + + def tag_resource(self) -> str: + arn = unquote(self.path.split("/tags/")[-1]) + tags = self._get_param("Tags") + self.appconfig_backend.tag_resource(arn, tags) + return "{}" + + def untag_resource(self) -> str: + arn = unquote(self.path.split("/tags/")[-1]) + tag_keys = self.querystring.get("tagKeys") + self.appconfig_backend.untag_resource(arn, tag_keys) # type: ignore[arg-type] + return "{}" + + def create_hosted_configuration_version(self) -> Tuple[str, Dict[str, Any]]: + app_id = self._get_param("ApplicationId") + config_profile_id = self._get_param("ConfigurationProfileId") + description = self.headers.get("Description") + content = self.body + content_type = self.headers.get("Content-Type") + version_label = self.headers.get("VersionLabel") + version = self.appconfig_backend.create_hosted_configuration_version( + app_id=app_id, + config_profile_id=config_profile_id, + description=description, + content=content, + content_type=content_type, + version_label=version_label, + ) + return version.content, version.get_headers() + + def get_hosted_configuration_version(self) -> Tuple[str, Dict[str, Any]]: + app_id = self._get_param("ApplicationId") + config_profile_id = self._get_param("ConfigurationProfileId") + version_number = self._get_int_param("VersionNumber") + version = self.appconfig_backend.get_hosted_configuration_version( + app_id=app_id, + config_profile_id=config_profile_id, + version=version_number, + ) + return version.content, version.get_headers() + + def delete_hosted_configuration_version(self) -> str: + app_id = self._get_param("ApplicationId") + config_profile_id = self._get_param("ConfigurationProfileId") + version_number = self._get_int_param("VersionNumber") + self.appconfig_backend.delete_hosted_configuration_version( + app_id=app_id, + config_profile_id=config_profile_id, + version=version_number, + ) + return "{}" diff --git a/moto/appconfig/urls.py b/moto/appconfig/urls.py new file mode 100644 index 000000000..560c14dee --- /dev/null +++ b/moto/appconfig/urls.py @@ -0,0 +1,22 @@ +"""appconfig base URL and path.""" +from .responses import AppConfigResponse + +url_bases = [ + r"https?://appconfig\.(.+)\.amazonaws\.com", +] + + +response = AppConfigResponse() + + +url_paths = { + "{0}/applications$": response.dispatch, + "{0}/applications/(?P[^/]+)$": response.dispatch, + "{0}/applications/(?P[^/]+)/configurationprofiles$": response.dispatch, + "{0}/applications/(?P[^/]+)/configurationprofiles/(?P[^/]+)$": response.dispatch, + "{0}/applications/(?P[^/]+)/configurationprofiles/(?P[^/]+)/hostedconfigurationversions$": response.dispatch, + "{0}/applications/(?P[^/]+)/configurationprofiles/(?P[^/]+)/hostedconfigurationversions/(?P[^/]+)$": response.dispatch, + "{0}/tags/(?P.+)$": response.dispatch, + "{0}/tags/(?P[^/]+)/(?P[^/]+)$": response.tags, + "{0}/tags/(?P[^/]+)/(?P[^/]+)/configurationprofile/(?P[^/]+)$": response.tags, +} diff --git a/moto/backend_index.py b/moto/backend_index.py index 7b10be586..9f99ed313 100644 --- a/moto/backend_index.py +++ b/moto/backend_index.py @@ -6,6 +6,7 @@ backend_url_patterns = [ ("acm-pca", re.compile("https?://acm-pca\\.(.+)\\.amazonaws\\.com")), ("amp", re.compile("https?://aps\\.(.+)\\.amazonaws\\.com")), ("apigateway", re.compile("https?://apigateway\\.(.+)\\.amazonaws.com")), + ("appconfig", re.compile("https?://appconfig\\.(.+)\\.amazonaws\\.com")), ( "applicationautoscaling", re.compile("https?://application-autoscaling\\.(.+)\\.amazonaws.com"), diff --git a/tests/terraformtests/terraform-tests.success.txt b/tests/terraformtests/terraform-tests.success.txt index 7011067bb..b23717eed 100644 --- a/tests/terraformtests/terraform-tests.success.txt +++ b/tests/terraformtests/terraform-tests.success.txt @@ -31,6 +31,12 @@ apigatewayv2: - TestAccAPIGatewayV2Model - TestAccAPIGatewayV2Route - TestAccAPIGatewayV2VPCLink +appconfig: + - TestAccAppConfigConfigurationProfileDataSource_basic + - TestAccAppConfigConfigurationProfile_ + - TestAccAppConfigConfigurationProfilesDataSource_basic + - TestAccAppConfigApplication_ + - TestAccAppConfigHostedConfigurationVersion_ autoscaling: - TestAccAutoScalingAttachment - TestAccAutoScalingGroupDataSource diff --git a/tests/test_appconfig/__init__.py b/tests/test_appconfig/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_appconfig/test_appconfig_applications.py b/tests/test_appconfig/test_appconfig_applications.py new file mode 100644 index 000000000..76dfe65df --- /dev/null +++ b/tests/test_appconfig/test_appconfig_applications.py @@ -0,0 +1,83 @@ +"""Unit tests for appconfig-supported APIs.""" +import boto3 +import pytest + +from botocore.exceptions import ClientError +from moto import mock_appconfig + +# See our Development Tips on writing tests for hints on how to write good tests: +# http://docs.getmoto.org/en/latest/docs/contributing/development_tips/tests.html + + +@mock_appconfig +def test_create_application(): + client = boto3.client("appconfig", region_name="ap-southeast-1") + post = client.create_application(Name="testapp", Description="blah") + + assert "Id" in post + assert post["Name"] == "testapp" + assert post["Description"] == "blah" + + get = client.get_application(ApplicationId=post["Id"]) + assert post["Id"] == get["Id"] + assert post["Name"] == get["Name"] + assert post["Description"] == get["Description"] + + update = client.update_application( + ApplicationId=post["Id"], + Name="name2", + Description="desc2", + ) + assert update["Name"] == "name2" + assert update["Description"] == "desc2" + + client.delete_application(ApplicationId=post["Id"]) + + with pytest.raises(ClientError) as exc: + client.get_application(ApplicationId=post["Id"]) + err = exc.value.response["Error"] + assert err["Code"] == "ResourceNotFoundException" + + +@mock_appconfig +def test_tag_application(): + client = boto3.client("appconfig", region_name="us-east-2") + app_id = client.create_application(Name="testapp")["Id"] + app_arn = f"arn:aws:appconfig:us-east-2:123456789012:application/{app_id}" + + tags = client.list_tags_for_resource(ResourceArn=app_arn)["Tags"] + assert tags == {} + + client.tag_resource( + ResourceArn=app_arn, + Tags={"k1": "v1"}, + ) + + tags = client.list_tags_for_resource(ResourceArn=app_arn)["Tags"] + assert tags == {"k1": "v1"} + + #### + # Check this flow works when creating an app with tags + app_id = client.create_application(Name="testapp", Tags={"k1": "v1"})["Id"] + app_arn = f"arn:aws:appconfig:us-east-2:123456789012:application/{app_id}" + + tags = client.list_tags_for_resource(ResourceArn=app_arn)["Tags"] + assert tags == {"k1": "v1"} + + client.tag_resource( + ResourceArn=app_arn, + Tags={"k2": "v2", "k3": "v3"}, + ) + + tags = client.list_tags_for_resource(ResourceArn=app_arn)["Tags"] + assert tags == {"k1": "v1", "k2": "v2", "k3": "v3"} + + client.untag_resource(ResourceArn=app_arn, TagKeys=["k2"]) + + tags = client.list_tags_for_resource(ResourceArn=app_arn)["Tags"] + assert tags == {"k1": "v1", "k3": "v3"} + + client.untag_resource(ResourceArn=app_arn, TagKeys=["k1", "k3"]) + + tags = client.list_tags_for_resource(ResourceArn=app_arn)["Tags"] + assert tags == {} diff --git a/tests/test_appconfig/test_appconfig_config_profiles.py b/tests/test_appconfig/test_appconfig_config_profiles.py new file mode 100644 index 000000000..c0fb8943f --- /dev/null +++ b/tests/test_appconfig/test_appconfig_config_profiles.py @@ -0,0 +1,134 @@ +import boto3 +import pytest + +from botocore.exceptions import ClientError +from moto import mock_appconfig + + +@mock_appconfig +def test_create_configuration_profile(): + client = boto3.client("appconfig", region_name="eu-north-1") + app_id = client.create_application(Name="testapp")["Id"] + resp = client.create_configuration_profile( + ApplicationId=app_id, + Name="config_name", + Description="desc", + LocationUri="luri", + RetrievalRoleArn="rrarn:rrarn:rrarn:rrarn", + Validators=[{"Type": "JSON", "Content": "c"}], + Type="freeform", + ) + del resp["ResponseMetadata"] + assert "Id" in resp + config_profile_id = resp.pop("Id") + + expected_response = { + "ApplicationId": app_id, + "Name": "config_name", + "Description": "desc", + "LocationUri": "luri", + "RetrievalRoleArn": "rrarn:rrarn:rrarn:rrarn", + "Validators": [{"Type": "JSON", "Content": "c"}], + "Type": "freeform", + } + + assert resp == expected_response + + resp = client.get_configuration_profile( + ApplicationId=app_id, + ConfigurationProfileId=config_profile_id, + ) + del resp["ResponseMetadata"] + assert "Id" in resp + resp.pop("Id") + assert resp == expected_response + + profiles = client.list_configuration_profiles(ApplicationId=app_id)["Items"] + assert profiles == [ + { + "ApplicationId": app_id, + "Id": config_profile_id, + "Name": "config_name", + "LocationUri": "luri", + "Type": "freeform", + } + ] + + update = client.update_configuration_profile( + ApplicationId=app_id, + ConfigurationProfileId=config_profile_id, + Name="name2", + Description="desc2", + RetrievalRoleArn="rrarn:rrarn:rrarn:222", + Validators=[], + ) + assert update["Name"] == "name2" + assert update["RetrievalRoleArn"] == "rrarn:rrarn:rrarn:222" + + client.delete_configuration_profile( + ApplicationId=app_id, + ConfigurationProfileId=config_profile_id, + ) + + with pytest.raises(ClientError) as exc: + client.get_configuration_profile( + ApplicationId=app_id, + ConfigurationProfileId=config_profile_id, + ) + err = exc.value.response["Error"] + assert err["Code"] == "ResourceNotFoundException" + + +@mock_appconfig +def test_tag_config_profile(): + client = boto3.client("appconfig", region_name="us-east-2") + app_id = client.create_application(Name="testapp")["Id"] + cp_id = client.create_configuration_profile( + ApplicationId=app_id, + Name="config_name", + LocationUri="luri", + )["Id"] + cp_arn = f"arn:aws:appconfig:us-east-2:123456789012:application/{app_id}/configurationprofile/{cp_id}" + + tags = client.list_tags_for_resource(ResourceArn=cp_arn)["Tags"] + assert tags == {} + + client.tag_resource( + ResourceArn=cp_arn, + Tags={"k1": "v1"}, + ) + + tags = client.list_tags_for_resource(ResourceArn=cp_arn)["Tags"] + assert tags == {"k1": "v1"} + + #### + # Check this flow works when creating an app with tags + app_id = client.create_application(Name="testapp")["Id"] + cp_id = client.create_configuration_profile( + ApplicationId=app_id, + Name="config_name", + LocationUri="luri", + Tags={"k1": "v1"}, + )["Id"] + cp_arn = f"arn:aws:appconfig:us-east-2:123456789012:application/{app_id}/configurationprofile/{cp_id}" + + tags = client.list_tags_for_resource(ResourceArn=cp_arn)["Tags"] + assert tags == {"k1": "v1"} + + client.tag_resource( + ResourceArn=cp_arn, + Tags={"k2": "v2", "k3": "v3"}, + ) + + tags = client.list_tags_for_resource(ResourceArn=cp_arn)["Tags"] + assert tags == {"k1": "v1", "k2": "v2", "k3": "v3"} + + client.untag_resource(ResourceArn=cp_arn, TagKeys=["k2"]) + + tags = client.list_tags_for_resource(ResourceArn=cp_arn)["Tags"] + assert tags == {"k1": "v1", "k3": "v3"} + + client.untag_resource(ResourceArn=cp_arn, TagKeys=["k1", "k3"]) + + tags = client.list_tags_for_resource(ResourceArn=cp_arn)["Tags"] + assert tags == {} diff --git a/tests/test_appconfig/test_appconfig_hosted_config_versions.py b/tests/test_appconfig/test_appconfig_hosted_config_versions.py new file mode 100644 index 000000000..40748f9f3 --- /dev/null +++ b/tests/test_appconfig/test_appconfig_hosted_config_versions.py @@ -0,0 +1,88 @@ +import boto3 +import pytest + +from botocore.exceptions import ClientError +from moto import mock_appconfig + + +@mock_appconfig +class TestHostedConfigurationVersions: + def setup_method(self, *args): # pylint: disable=unused-argument + self.client = boto3.client("appconfig", region_name="us-west-1") + self.app_id = self.client.create_application(Name="testapp")["Id"] + self.config_profile_id = self.client.create_configuration_profile( + ApplicationId=self.app_id, + Name="config_name", + Description="desc", + LocationUri="luri", + RetrievalRoleArn="rrarn:rrarn:rrarn:rrarn", + Validators=[{"Type": "JSON", "Content": "c"}], + Type="freeform", + )["Id"] + + def test_create_hosted_configuration_version(self): + resp = self.client.create_hosted_configuration_version( + ApplicationId=self.app_id, + ConfigurationProfileId=self.config_profile_id, + Description="desc", + Content=b"asdf", + ContentType="text/xml", + VersionLabel="vl", + ) + assert resp["ApplicationId"] == self.app_id + assert resp["ConfigurationProfileId"] == self.config_profile_id + assert resp["VersionNumber"] == 1 + assert resp["Description"] == "desc" + assert resp["VersionLabel"] == "vl" + assert resp["ContentType"] == "text/xml" + assert resp["Content"].read() == b"asdf" + + resp = self.client.create_hosted_configuration_version( + ApplicationId=self.app_id, + ConfigurationProfileId=self.config_profile_id, + Content=b"asdf", + ContentType="text/xml", + ) + assert resp["VersionNumber"] == 2 + + def test_get_hosted_configuration_version(self): + self.client.create_hosted_configuration_version( + ApplicationId=self.app_id, + ConfigurationProfileId=self.config_profile_id, + Description="desc", + Content=b"asdf", + ContentType="text/xml", + VersionLabel="vl", + ) + get = self.client.get_hosted_configuration_version( + ApplicationId=self.app_id, + ConfigurationProfileId=self.config_profile_id, + VersionNumber=1, + ) + assert get["ApplicationId"] == self.app_id + assert get["ConfigurationProfileId"] == self.config_profile_id + assert get["Description"] == "desc" + assert get["VersionLabel"] == "vl" + assert get["ContentType"] == "text/xml" + assert get["Content"].read() == b"asdf" + + def test_delete_hosted_configuration_version(self): + self.client.create_hosted_configuration_version( + ApplicationId=self.app_id, + ConfigurationProfileId=self.config_profile_id, + Content=b"asdf", + ContentType="text/xml", + ) + self.client.delete_hosted_configuration_version( + ApplicationId=self.app_id, + ConfigurationProfileId=self.config_profile_id, + VersionNumber=1, + ) + with pytest.raises(ClientError) as exc: + self.client.get_hosted_configuration_version( + ApplicationId=self.app_id, + ConfigurationProfileId=self.config_profile_id, + VersionNumber=1, + ) + err = exc.value.response["Error"] + assert err["Code"] == "ResourceNotFoundException"