Feature: AppConfig (#6391)

This commit is contained in:
Bert Blommers 2023-06-11 14:45:26 +00:00 committed by GitHub
parent 6d9f4646c1
commit 6b86113534
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 935 additions and 1 deletions

View File

@ -280,6 +280,55 @@
- [X] update_vpc_link
</details>
## appconfig
<details>
<summary>34% implemented</summary>
- [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
</details>
## application-autoscaling
<details>
<summary>69% implemented</summary>
@ -7121,7 +7170,6 @@
- amplifybackend
- amplifyuibuilder
- apigatewaymanagementapi
- appconfig
- appconfigdata
- appflow
- appintegrations

View File

@ -0,0 +1,77 @@
.. _implementedservice_appconfig:
.. |start-h3| raw:: html
<h3>
.. |end-h3| raw:: html
</h3>
=========
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

View File

@ -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(

View File

@ -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)

View File

@ -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"
)

281
moto/appconfig/models.py Normal file
View File

@ -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")

169
moto/appconfig/responses.py Normal file
View File

@ -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 "{}"

22
moto/appconfig/urls.py Normal file
View File

@ -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<app_id>[^/]+)$": response.dispatch,
"{0}/applications/(?P<app_id>[^/]+)/configurationprofiles$": response.dispatch,
"{0}/applications/(?P<app_id>[^/]+)/configurationprofiles/(?P<config_profile_id>[^/]+)$": response.dispatch,
"{0}/applications/(?P<app_id>[^/]+)/configurationprofiles/(?P<config_profile_id>[^/]+)/hostedconfigurationversions$": response.dispatch,
"{0}/applications/(?P<app_id>[^/]+)/configurationprofiles/(?P<config_profile_id>[^/]+)/hostedconfigurationversions/(?P<version>[^/]+)$": response.dispatch,
"{0}/tags/(?P<app_id>.+)$": response.dispatch,
"{0}/tags/(?P<arn_part_1>[^/]+)/(?P<app_id>[^/]+)$": response.tags,
"{0}/tags/(?P<arn_part_1>[^/]+)/(?P<app_id>[^/]+)/configurationprofile/(?P<cp_id>[^/]+)$": response.tags,
}

View File

@ -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"),

View File

@ -31,6 +31,12 @@ apigatewayv2:
- TestAccAPIGatewayV2Model
- TestAccAPIGatewayV2Route
- TestAccAPIGatewayV2VPCLink
appconfig:
- TestAccAppConfigConfigurationProfileDataSource_basic
- TestAccAppConfigConfigurationProfile_
- TestAccAppConfigConfigurationProfilesDataSource_basic
- TestAccAppConfigApplication_
- TestAccAppConfigHostedConfigurationVersion_
autoscaling:
- TestAccAutoScalingAttachment
- TestAccAutoScalingGroupDataSource

View File

View File

@ -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 == {}

View File

@ -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 == {}

View File

@ -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"