diff --git a/IMPLEMENTATION_COVERAGE.md b/IMPLEMENTATION_COVERAGE.md index 1e1db0b3d..fd1fc58d2 100644 --- a/IMPLEMENTATION_COVERAGE.md +++ b/IMPLEMENTATION_COVERAGE.md @@ -20,6 +20,33 @@ - [ ] update_certificate_options +## amp +
+61% implemented + +- [ ] create_alert_manager_definition +- [ ] create_logging_configuration +- [X] create_rule_groups_namespace +- [X] create_workspace +- [ ] delete_alert_manager_definition +- [ ] delete_logging_configuration +- [X] delete_rule_groups_namespace +- [X] delete_workspace +- [ ] describe_alert_manager_definition +- [ ] describe_logging_configuration +- [X] describe_rule_groups_namespace +- [X] describe_workspace +- [X] list_rule_groups_namespaces +- [X] list_tags_for_resource +- [X] list_workspaces +- [ ] put_alert_manager_definition +- [X] put_rule_groups_namespace +- [X] tag_resource +- [X] untag_resource +- [ ] update_logging_configuration +- [X] update_workspace_alias +
+ ## apigateway
65% implemented @@ -6224,7 +6251,6 @@ - account - acm-pca - alexaforbusiness -- amp - amplify - amplifybackend - amplifyuibuilder diff --git a/docs/docs/services/amp.rst b/docs/docs/services/amp.rst new file mode 100644 index 000000000..7dec8bd6e --- /dev/null +++ b/docs/docs/services/amp.rst @@ -0,0 +1,75 @@ +.. _implementedservice_amp: + +.. |start-h3| raw:: html + +

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

+ +=== +amp +=== + +.. autoclass:: moto.amp.models.PrometheusServiceBackend + +|start-h3| Example usage |end-h3| + +.. sourcecode:: python + + @mock_amp + def test_amp_behaviour: + boto3.client("amp") + ... + + + +|start-h3| Implemented features for this service |end-h3| + +- [ ] create_alert_manager_definition +- [ ] create_logging_configuration +- [X] create_rule_groups_namespace + + The ClientToken-parameter is not yet implemented + + +- [X] create_workspace + + The ClientToken-parameter is not yet implemented + + +- [ ] delete_alert_manager_definition +- [ ] delete_logging_configuration +- [X] delete_rule_groups_namespace + + The ClientToken-parameter is not yet implemented + + +- [X] delete_workspace + + The ClientToken-parameter is not yet implemented + + +- [ ] describe_alert_manager_definition +- [ ] describe_logging_configuration +- [X] describe_rule_groups_namespace +- [X] describe_workspace +- [X] list_rule_groups_namespaces +- [X] list_tags_for_resource +- [X] list_workspaces +- [ ] put_alert_manager_definition +- [X] put_rule_groups_namespace + + The ClientToken-parameter is not yet implemented + + +- [X] tag_resource +- [X] untag_resource +- [ ] update_logging_configuration +- [X] update_workspace_alias + + The ClientToken-parameter is not yet implemented + + + diff --git a/moto/__init__.py b/moto/__init__.py index a5d798304..b9f0dda36 100644 --- a/moto/__init__.py +++ b/moto/__init__.py @@ -16,6 +16,7 @@ def lazy_load(module_name, element, boto3_name=None, backend=None): mock_acm = lazy_load(".acm", "mock_acm") +mock_amp = lazy_load(".amp", "mock_amp") mock_apigateway = lazy_load(".apigateway", "mock_apigateway") mock_apigatewayv2 = lazy_load(".apigatewayv2", "mock_apigatewayv2") mock_appsync = lazy_load(".appsync", "mock_appsync") diff --git a/moto/amp/__init__.py b/moto/amp/__init__.py new file mode 100644 index 000000000..a135ba7d5 --- /dev/null +++ b/moto/amp/__init__.py @@ -0,0 +1,5 @@ +"""amp module initialization; sets value for base decorator.""" +from .models import amp_backends +from ..core.models import base_decorator + +mock_amp = base_decorator(amp_backends) diff --git a/moto/amp/exceptions.py b/moto/amp/exceptions.py new file mode 100644 index 000000000..45e8e789f --- /dev/null +++ b/moto/amp/exceptions.py @@ -0,0 +1,40 @@ +import json +from moto.core.exceptions import JsonRESTError + + +class AmpException(JsonRESTError): + pass + + +class ResourceNotFoundException(AmpException): + def __init__(self, message, resource_id, resource_type): + super().__init__("ResourceNotFoundException", message) + self.description = json.dumps( + { + "resourceId": resource_id, + "message": self.message, + "resourceType": resource_type, + } + ) + + +class WorkspaceNotFound(ResourceNotFoundException): + code = 404 + + def __init__(self, workspace_id): + super().__init__( + "Workspace not found", + resource_id=workspace_id, + resource_type="AWS::APS::Workspace", + ) + + +class RuleGroupNamespaceNotFound(ResourceNotFoundException): + code = 404 + + def __init__(self, name): + super().__init__( + "RuleGroupNamespace not found", + resource_id=name, + resource_type="AWS::APS::RuleGroupNamespace", + ) diff --git a/moto/amp/models.py b/moto/amp/models.py new file mode 100644 index 000000000..1159d62fb --- /dev/null +++ b/moto/amp/models.py @@ -0,0 +1,168 @@ +"""PrometheusServiceBackend class with methods for supported APIs.""" + +from moto.core import BaseBackend, BaseModel +from moto.core.utils import BackendDict, unix_time +from moto.utilities.paginator import paginate +from moto.utilities.tagging_service import TaggingService +from typing import Dict +from uuid import uuid4 +from .exceptions import RuleGroupNamespaceNotFound, WorkspaceNotFound +from .utils import PAGINATION_MODEL + + +class RuleGroupNamespace(BaseModel): + def __init__(self, account_id, region, workspace_id, name, data, tag_fn): + self.name = name + self.data = data + self.tag_fn = tag_fn + self.arn = f"arn:aws:aps:{region}:{account_id}:rulegroupsnamespace/{workspace_id}/{self.name}" + self.created_at = unix_time() + self.modified_at = self.created_at + + def update(self, new_data): + self.data = new_data + self.modified_at = unix_time() + + def to_dict(self): + return { + "name": self.name, + "arn": self.arn, + "status": {"statusCode": "ACTIVE"}, + "createdAt": self.created_at, + "modifiedAt": self.modified_at, + "data": self.data, + "tags": self.tag_fn(self.arn), + } + + +class Workspace(BaseModel): + def __init__(self, account_id, region, alias, tag_fn): + self.alias = alias + self.workspace_id = f"ws-{uuid4()}" + self.arn = f"arn:aws:aps:{region}:{account_id}:workspace/{self.workspace_id}" + self.endpoint = f"https://aps-workspaces.{region}.amazonaws.com/workspaces/{self.workspace_id}/" + self.status = {"statusCode": "ACTIVE"} + self.created_at = unix_time() + self.tag_fn = tag_fn + self.rule_group_namespaces = dict() + + def to_dict(self): + return { + "alias": self.alias, + "arn": self.arn, + "workspaceId": self.workspace_id, + "status": self.status, + "createdAt": self.created_at, + "prometheusEndpoint": self.endpoint, + "tags": self.tag_fn(self.arn), + } + + +class PrometheusServiceBackend(BaseBackend): + """Implementation of PrometheusService APIs.""" + + def __init__(self, region_name, account_id): + super().__init__(region_name, account_id) + self.workspaces: Dict(str, Workspace) = dict() + self.tagger = TaggingService() + + def create_workspace(self, alias, tags): + """ + The ClientToken-parameter is not yet implemented + """ + workspace = Workspace( + self.account_id, + self.region_name, + alias=alias, + tag_fn=self.list_tags_for_resource, + ) + self.workspaces[workspace.workspace_id] = workspace + self.tag_resource(workspace.arn, tags) + return workspace + + def describe_workspace(self, workspace_id) -> Workspace: + if workspace_id not in self.workspaces: + raise WorkspaceNotFound(workspace_id) + return self.workspaces[workspace_id] + + def list_tags_for_resource(self, resource_arn): + return self.tagger.get_tag_dict_for_resource(resource_arn) + + def update_workspace_alias(self, alias, workspace_id): + """ + The ClientToken-parameter is not yet implemented + """ + self.workspaces[workspace_id].alias = alias + + def delete_workspace(self, workspace_id): + """ + The ClientToken-parameter is not yet implemented + """ + self.workspaces.pop(workspace_id, None) + + @paginate(pagination_model=PAGINATION_MODEL) + def list_workspaces(self, alias): + if alias: + return [w for w in self.workspaces.values() if w.alias == alias] + return list(self.workspaces.values()) + + def tag_resource(self, resource_arn, tags): + tags = self.tagger.convert_dict_to_tags_input(tags) + self.tagger.tag_resource(resource_arn, tags) + + def untag_resource(self, resource_arn, tag_keys): + self.tagger.untag_resource_using_names(resource_arn, tag_keys) + + def create_rule_groups_namespace( + self, data, name, tags, workspace_id + ) -> RuleGroupNamespace: + """ + The ClientToken-parameter is not yet implemented + """ + workspace = self.describe_workspace(workspace_id) + group = RuleGroupNamespace( + account_id=self.account_id, + region=self.region_name, + workspace_id=workspace_id, + name=name, + data=data, + tag_fn=self.list_tags_for_resource, + ) + workspace.rule_group_namespaces[name] = group + self.tag_resource(group.arn, tags) + return group + + def delete_rule_groups_namespace(self, name, workspace_id) -> None: + """ + The ClientToken-parameter is not yet implemented + """ + ws = self.describe_workspace(workspace_id) + ws.rule_group_namespaces.pop(name, None) + + def describe_rule_groups_namespace(self, name, workspace_id) -> RuleGroupNamespace: + ws = self.describe_workspace(workspace_id) + if name not in ws.rule_group_namespaces: + raise RuleGroupNamespaceNotFound(name=name) + return ws.rule_group_namespaces[name] + + def put_rule_groups_namespace(self, data, name, workspace_id) -> RuleGroupNamespace: + """ + The ClientToken-parameter is not yet implemented + """ + ns = self.describe_rule_groups_namespace(name=name, workspace_id=workspace_id) + ns.update(data) + return ns + + @paginate(pagination_model=PAGINATION_MODEL) + def list_rule_groups_namespaces(self, name, workspace_id): + ws = self.describe_workspace(workspace_id) + if name: + return [ + ns + for ns_name, ns in ws.rule_group_namespaces.items() + if ns_name.startswith(name) + ] + return list(ws.rule_group_namespaces.values()) + + +amp_backends = BackendDict(PrometheusServiceBackend, "amp") diff --git a/moto/amp/responses.py b/moto/amp/responses.py new file mode 100644 index 000000000..ec1995ce7 --- /dev/null +++ b/moto/amp/responses.py @@ -0,0 +1,141 @@ +"""Handles incoming amp requests, invokes methods, returns responses.""" +import json + +from moto.core.responses import BaseResponse +from .models import amp_backends, PrometheusServiceBackend +from urllib.parse import unquote + + +class PrometheusServiceResponse(BaseResponse): + """Handler for PrometheusService requests and responses.""" + + def tags(self, request, full_url, headers): + 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): + super().__init__(service_name="amp") + + @property + def amp_backend(self) -> PrometheusServiceBackend: + """Return backend instance specific for this region.""" + return amp_backends[self.current_account][self.region] + + def create_workspace(self): + params = json.loads(self.body) + alias = params.get("alias") + tags = params.get("tags") + workspace = self.amp_backend.create_workspace(alias=alias, tags=tags) + return json.dumps(dict(workspace.to_dict())) + + def describe_workspace(self): + workspace_id = self.path.split("/")[-1] + workspace = self.amp_backend.describe_workspace(workspace_id=workspace_id) + return json.dumps(dict(workspace=workspace.to_dict())) + + def list_tags_for_resource(self): + resource_arn = unquote(self.path).split("tags/")[-1] + tags = self.amp_backend.list_tags_for_resource(resource_arn=resource_arn) + return json.dumps(dict(tags=tags)) + + def update_workspace_alias(self): + params = json.loads(self.body) + alias = params.get("alias") + workspace_id = self.path.split("/")[-2] + self.amp_backend.update_workspace_alias(alias=alias, workspace_id=workspace_id) + return json.dumps(dict()) + + def delete_workspace(self): + workspace_id = self.path.split("/")[-1] + self.amp_backend.delete_workspace(workspace_id=workspace_id) + return json.dumps(dict()) + + def list_workspaces(self): + alias = self._get_param("alias") + max_results = self._get_int_param("maxResults") + next_token = self._get_param("nextToken") + workspaces, next_token = self.amp_backend.list_workspaces( + alias, max_results=max_results, next_token=next_token + ) + return json.dumps( + {"nextToken": next_token, "workspaces": [w.to_dict() for w in workspaces]} + ) + + def tag_resource(self): + params = json.loads(self.body) + resource_arn = unquote(self.path).split("tags/")[-1] + tags = params.get("tags") + self.amp_backend.tag_resource(resource_arn=resource_arn, tags=tags) + return json.dumps(dict()) + + def untag_resource(self): + resource_arn = unquote(self.path).split("tags/")[-1] + tag_keys = self.querystring.get("tagKeys", []) + self.amp_backend.untag_resource(resource_arn=resource_arn, tag_keys=tag_keys) + return json.dumps(dict()) + + def create_rule_groups_namespace(self): + params = json.loads(self.body) + data = params.get("data") + name = params.get("name") + tags = params.get("tags") + workspace_id = unquote(self.path).split("/")[-2] + rule_group_namespace = self.amp_backend.create_rule_groups_namespace( + data=data, + name=name, + tags=tags, + workspace_id=workspace_id, + ) + return json.dumps(rule_group_namespace.to_dict()) + + def delete_rule_groups_namespace(self): + name = unquote(self.path).split("/")[-1] + workspace_id = unquote(self.path).split("/")[-3] + self.amp_backend.delete_rule_groups_namespace( + name=name, + workspace_id=workspace_id, + ) + return json.dumps(dict()) + + def describe_rule_groups_namespace(self): + name = unquote(self.path).split("/")[-1] + workspace_id = unquote(self.path).split("/")[-3] + ns = self.amp_backend.describe_rule_groups_namespace( + name=name, workspace_id=workspace_id + ) + return json.dumps(dict(ruleGroupsNamespace=ns.to_dict())) + + def put_rule_groups_namespace(self): + params = json.loads(self.body) + data = params.get("data") + name = unquote(self.path).split("/")[-1] + workspace_id = unquote(self.path).split("/")[-3] + ns = self.amp_backend.put_rule_groups_namespace( + data=data, + name=name, + workspace_id=workspace_id, + ) + return json.dumps(ns.to_dict()) + + def list_rule_groups_namespaces(self): + max_results = self._get_int_param("maxResults") + next_token = self._get_param("nextToken") + name = self._get_param("name") + workspace_id = unquote(self.path).split("/")[-2] + namespaces, next_token = self.amp_backend.list_rule_groups_namespaces( + max_results=max_results, + name=name, + next_token=next_token, + workspace_id=workspace_id, + ) + return json.dumps( + dict( + nextToken=next_token, + ruleGroupsNamespaces=[ns.to_dict() for ns in namespaces], + ) + ) diff --git a/moto/amp/urls.py b/moto/amp/urls.py new file mode 100644 index 000000000..b18f717a7 --- /dev/null +++ b/moto/amp/urls.py @@ -0,0 +1,21 @@ +"""amp base URL and path.""" +from .responses import PrometheusServiceResponse + +url_bases = [ + r"https?://aps\.(.+)\.amazonaws\.com", +] + + +response = PrometheusServiceResponse() + + +url_paths = { + "{0}/workspaces$": response.dispatch, + "{0}/workspaces/(?P[^/]+)$": response.dispatch, + "{0}/workspaces/(?P[^/]+)/alias$": response.dispatch, + "{0}/workspaces/(?P[^/]+)/rulegroupsnamespaces$": response.dispatch, + "{0}/workspaces/(?P[^/]+)/rulegroupsnamespaces/(?P[^/]+)$": response.dispatch, + "{0}/tags/(?P[^/]+)$": response.dispatch, + "{0}/tags/(?P[^/]+)/(?P[^/]+)$": response.tags, + "{0}/tags/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)$": response.tags, +} diff --git a/moto/amp/utils.py b/moto/amp/utils.py new file mode 100644 index 000000000..4d548f87a --- /dev/null +++ b/moto/amp/utils.py @@ -0,0 +1,14 @@ +PAGINATION_MODEL = { + "list_workspaces": { + "input_token": "next_token", + "limit_key": "max_results", + "limit_default": 100, # This should be the sum of the directory limits + "unique_attribute": "arn", + }, + "list_rule_groups_namespaces": { + "input_token": "next_token", + "limit_key": "max_results", + "limit_default": 100, # This should be the sum of the directory limits + "unique_attribute": "name", + }, +} diff --git a/moto/backend_index.py b/moto/backend_index.py index 6b6c934e2..df258077e 100644 --- a/moto/backend_index.py +++ b/moto/backend_index.py @@ -3,6 +3,7 @@ import re backend_url_patterns = [ ("acm", re.compile("https?://acm\\.(.+)\\.amazonaws\\.com")), + ("amp", re.compile("https?://aps\\.(.+)\\.amazonaws\\.com")), ("apigateway", re.compile("https?://apigateway\\.(.+)\\.amazonaws.com")), ( "applicationautoscaling", diff --git a/tests/terraformtests/terraform-tests.success.txt b/tests/terraformtests/terraform-tests.success.txt index 6a2aaa3d1..8fae5fadf 100644 --- a/tests/terraformtests/terraform-tests.success.txt +++ b/tests/terraformtests/terraform-tests.success.txt @@ -1,5 +1,8 @@ acm: - TestAccACMCertificateDataSource +amp: + - TestAccAMPWorkspace + - TestAccAMPRuleGroupNamespace apigateway: - TestAccAPIGatewayGatewayResponse apigatewayv2: diff --git a/tests/test_amp/__init__.py b/tests/test_amp/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/test_amp/test_amp_rulegroupnamespaces.py b/tests/test_amp/test_amp_rulegroupnamespaces.py new file mode 100644 index 000000000..7218f26a5 --- /dev/null +++ b/tests/test_amp/test_amp_rulegroupnamespaces.py @@ -0,0 +1,173 @@ +"""Unit tests for amp-supported APIs.""" +import boto3 +import pytest +import sure # noqa # pylint: disable=unused-import + +from botocore.exceptions import ClientError +from moto import mock_amp + +# 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_amp +def test_create_rule_groups_namespace(): + client = boto3.client("amp", region_name="ap-southeast-1") + workspace_id = client.create_workspace()["workspaceId"] + resp = client.create_rule_groups_namespace( + data=b"asdf", name="my first rule group", workspaceId=workspace_id + ) + + resp.should.have.key("arn") + resp.should.have.key("name").equals("my first rule group") + resp.should.have.key("status") + + +@mock_amp +def test_delete_rule_groups_namespace(): + client = boto3.client("amp", region_name="us-east-2") + workspace_id = client.create_workspace()["workspaceId"] + client.create_rule_groups_namespace( + data=b"asdf", name="myname", workspaceId=workspace_id + ) + + client.delete_rule_groups_namespace(name="myname", workspaceId=workspace_id) + + with pytest.raises(ClientError) as exc: + client.describe_rule_groups_namespace(name="myname", workspaceId=workspace_id) + err = exc.value.response["Error"] + err["Code"].should.equal("ResourceNotFoundException") + err["Message"].should.equal("RuleGroupNamespace not found") + + +@mock_amp +def test_describe_rule_groups_namespace(): + client = boto3.client("amp", region_name="us-east-2") + + workspace_id = client.create_workspace()["workspaceId"] + client.create_rule_groups_namespace( + data=b"asdf", name="myname", workspaceId=workspace_id + ) + + resp = client.describe_rule_groups_namespace( + name="myname", workspaceId=workspace_id + ) + resp.should.have.key("ruleGroupsNamespace") + ns = resp["ruleGroupsNamespace"] + + ns.should.have.key("arn") + ns.should.have.key("createdAt") + ns.should.have.key("data").equals(b"asdf") + ns.should.have.key("modifiedAt") + ns.should.have.key("name").equals("myname") + ns.should.have.key("status") + + +@mock_amp +def test_put_rule_groups_namespace(): + client = boto3.client("amp", region_name="eu-west-1") + + workspace_id = client.create_workspace()["workspaceId"] + client.create_rule_groups_namespace( + data=b"asdf", name="myname", workspaceId=workspace_id + ) + + client.put_rule_groups_namespace( + name="myname", workspaceId=workspace_id, data=b"updated" + ) + + resp = client.describe_rule_groups_namespace( + name="myname", workspaceId=workspace_id + ) + resp.should.have.key("ruleGroupsNamespace") + ns = resp["ruleGroupsNamespace"] + + ns.should.have.key("arn") + ns.should.have.key("createdAt") + ns.should.have.key("data").equals(b"updated") + + +@mock_amp +def test_list_rule_groups_namespaces(): + client = boto3.client("amp", region_name="ap-southeast-1") + w_id = client.create_workspace()["workspaceId"] + for idx in range(15): + client.create_rule_groups_namespace( + data=b"a", name=f"ns{idx}", workspaceId=w_id + ) + + resp = client.list_rule_groups_namespaces(workspaceId=w_id) + resp.should.have.key("ruleGroupsNamespaces").length_of(15) + resp.shouldnt.have.key("nextToken") + + resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns1") + resp.should.have.key("ruleGroupsNamespaces").length_of(6) + names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]] + set(names).should.equal({"ns10", "ns13", "ns1", "ns12", "ns11", "ns14"}) + + resp = client.list_rule_groups_namespaces(workspaceId=w_id, name="ns10") + resp.should.have.key("ruleGroupsNamespaces").length_of(1) + names = [ns["name"] for ns in resp["ruleGroupsNamespaces"]] + set(names).should.equal({"ns10"}) + + +@mock_amp +def test_list_rule_groups_namespaces__paginated(): + client = boto3.client("amp", region_name="ap-southeast-1") + w_id = client.create_workspace()["workspaceId"] + for idx in range(125): + client.create_rule_groups_namespace( + data=b"a", name=f"ns{idx}", workspaceId=w_id + ) + + # default pagesize is 100 + page1 = client.list_rule_groups_namespaces(workspaceId=w_id) + page1.should.have.key("ruleGroupsNamespaces").length_of(100) + page1.should.have.key("nextToken") + + # We can ask for a smaller pagesize + page2 = client.list_rule_groups_namespaces( + workspaceId=w_id, maxResults=15, nextToken=page1["nextToken"] + ) + page2.should.have.key("ruleGroupsNamespaces").length_of(15) + page2.should.have.key("nextToken") + + page3 = client.list_rule_groups_namespaces( + workspaceId=w_id, maxResults=15, nextToken=page2["nextToken"] + ) + page3.should.have.key("ruleGroupsNamespaces").length_of(10) + page3.shouldnt.have.key("nextToken") + + # We could request all of them in one go + full_page = client.list_rule_groups_namespaces(workspaceId=w_id, maxResults=150) + full_page.should.have.key("ruleGroupsNamespaces").length_of(125) + full_page.shouldnt.have.key("nextToken") + + +@mock_amp +def test_tag_resource(): + client = boto3.client("amp", region_name="us-east-2") + + w_id = client.create_workspace()["workspaceId"] + ns = client.create_rule_groups_namespace( + data=b"a", name="ns", workspaceId=w_id, tags={"t": "v"} + ) + + arn = ns["arn"] + + client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"}) + + client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( + {"t": "v", "t1": "v1", "t2": "v2"} + ) + client.describe_rule_groups_namespace(workspaceId=w_id, name="ns")[ + "ruleGroupsNamespace" + ]["tags"].should.equal({"t": "v", "t1": "v1", "t2": "v2"}) + + client.untag_resource(resourceArn=arn, tagKeys=["t1"]) + client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( + {"t": "v", "t2": "v2"} + ) + + client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"]) + client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal({}) diff --git a/tests/test_amp/test_amp_workspaces.py b/tests/test_amp/test_amp_workspaces.py new file mode 100644 index 000000000..200e361f8 --- /dev/null +++ b/tests/test_amp/test_amp_workspaces.py @@ -0,0 +1,149 @@ +import boto3 +import pytest +import sure # noqa # pylint: disable=unused-import + +from botocore.exceptions import ClientError +from moto import mock_amp + +# 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_amp +def test_create_workspace(): + client = boto3.client("amp", region_name="ap-southeast-1") + resp = client.create_workspace(alias="test", clientToken="mytoken") + + resp.should.have.key("arn") + resp.should.have.key("status").equals({"statusCode": "ACTIVE"}) + resp.should.have.key("workspaceId") + + +@mock_amp +def test_describe_workspace(): + client = boto3.client("amp", region_name="eu-west-1") + workspace_id = client.create_workspace(alias="test", clientToken="mytoken")[ + "workspaceId" + ] + + resp = client.describe_workspace(workspaceId=workspace_id) + resp.should.have.key("workspace") + + workspace = resp["workspace"] + workspace.should.have.key("alias") + workspace.should.have.key("arn") + workspace.should.have.key("createdAt") + workspace.should.have.key("prometheusEndpoint") + workspace.should.have.key("status").equals({"statusCode": "ACTIVE"}) + workspace.should.have.key("workspaceId").equals(workspace_id) + + +@mock_amp +def test_list_workspaces(): + client = boto3.client("amp", region_name="ap-southeast-1") + client.create_workspace(alias="test") + client.create_workspace(alias="another") + client.create_workspace() + + resp = client.list_workspaces() + resp.should.have.key("workspaces").length_of(3) + resp.shouldnt.have.key("nextToken") + + resp = client.list_workspaces(alias="another") + resp.should.have.key("workspaces").length_of(1) + resp["workspaces"][0].should.have.key("alias").equals("another") + + +@mock_amp +def test_list_workspaces__paginated(): + client = boto3.client("amp", region_name="ap-southeast-1") + for _ in range(125): + client.create_workspace() + + # default pagesize is 100 + page1 = client.list_workspaces() + page1.should.have.key("workspaces").length_of(100) + page1.should.have.key("nextToken") + + # We can ask for a smaller pagesize + page2 = client.list_workspaces(maxResults=15, nextToken=page1["nextToken"]) + page2.should.have.key("workspaces").length_of(15) + page2.should.have.key("nextToken") + + page3 = client.list_workspaces(maxResults=15, nextToken=page2["nextToken"]) + page3.should.have.key("workspaces").length_of(10) + page3.shouldnt.have.key("nextToken") + + # We could request all of them in one go + full_page = client.list_workspaces(maxResults=150) + full_page.should.have.key("workspaces").length_of(125) + full_page.shouldnt.have.key("nextToken") + + +@mock_amp +def test_list_tags_for_resource(): + client = boto3.client("amp", region_name="ap-southeast-1") + arn = client.create_workspace( + alias="test", clientToken="mytoken", tags={"t1": "v1", "t2": "v2"} + )["arn"] + + resp = client.list_tags_for_resource(resourceArn=arn) + resp.should.have.key("tags").equals({"t1": "v1", "t2": "v2"}) + + +@mock_amp +def test_update_workspace_alias(): + client = boto3.client("amp", region_name="ap-southeast-1") + + workspace_id = client.create_workspace(alias="initial")["workspaceId"] + + w = client.describe_workspace(workspaceId=workspace_id)["workspace"] + w.should.have.key("alias").equals("initial") + + client.update_workspace_alias(alias="updated", workspaceId=workspace_id) + + w = client.describe_workspace(workspaceId=workspace_id)["workspace"] + w.should.have.key("alias").equals("updated") + + +@mock_amp +def test_delete_workspace(): + client = boto3.client("amp", region_name="us-east-2") + + workspace_id = client.create_workspace(alias="test", clientToken="mytoken")[ + "workspaceId" + ] + + client.delete_workspace(workspaceId=workspace_id) + + with pytest.raises(ClientError) as exc: + client.describe_workspace(workspaceId=workspace_id) + err = exc.value.response["Error"] + err["Code"].should.equal("ResourceNotFoundException") + err["Message"].should.equal("Workspace not found") + + +@mock_amp +def test_tag_resource(): + client = boto3.client("amp", region_name="us-east-2") + + workspace = client.create_workspace(alias="test", tags={"t": "v"}) + arn = workspace["arn"] + workspace_id = workspace["workspaceId"] + + client.tag_resource(resourceArn=arn, tags={"t1": "v1", "t2": "v2"}) + + client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( + {"t": "v", "t1": "v1", "t2": "v2"} + ) + client.describe_workspace(workspaceId=workspace_id)["workspace"][ + "tags" + ].should.equal({"t": "v", "t1": "v1", "t2": "v2"}) + + client.untag_resource(resourceArn=arn, tagKeys=["t1"]) + client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal( + {"t": "v", "t2": "v2"} + ) + + client.untag_resource(resourceArn=arn, tagKeys=["t", "t2"]) + client.list_tags_for_resource(resourceArn=arn)["tags"].should.equal({})