Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add SWF domain and type undeprecation

Browse Source 
Signed-off-by: Laurie O <laurie_opperman@hotmail.com>
This commit is contained in:
Laurie O 2020-03-05 23:37:17 +10:00
parent f8af496445
commit 916add9ac5
No known key found for this signature in database
GPG Key ID: AAA23A02196FC956
5 changed files with 235 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
moto/swf/models/__init__.py
View File

@ -121,6 +121,12 @@ class SWFBackend(BaseBackend):
raise SWFDomainDeprecatedFault(name)
domain.status = "DEPRECATED"
def undeprecate_domain(self, name):
domain = self._get_domain(name)
if domain.status == "REGISTERED":
raise SWFDomainAlreadyExistsFault(name)
domain.status = "REGISTERED"
def describe_domain(self, name):
return self._get_domain(name)
@ -148,6 +154,13 @@ class SWFBackend(BaseBackend):
raise SWFTypeDeprecatedFault(_type)
_type.status = "DEPRECATED"
def undeprecate_type(self, kind, domain_name, name, version):
domain = self._get_domain(domain_name)
_type = domain.get_type(kind, name, version)
if _type.status == "REGISTERED":
raise SWFTypeAlreadyExistsFault(_type)
_type.status = "REGISTERED"
def describe_type(self, kind, domain_name, name, version):
domain = self._get_domain(domain_name)
return domain.get_type(kind, name, version)

23
moto/swf/responses.py
View File

@ -92,6 +92,17 @@ class SWFResponse(BaseResponse):
self.swf_backend.deprecate_type(kind, domain, name, version)
return ""
def _undeprecate_type(self, kind):
domain = self._params["domain"]
_type_args = self._params["{0}Type".format(kind)]
name = _type_args["name"]
version = _type_args["version"]
self._check_string(domain)
self._check_string(name)
self._check_string(version)
self.swf_backend.undeprecate_type(kind, domain, name, version)
return ""
# TODO: implement pagination
def list_domains(self):
status = self._params["registrationStatus"]
@ -219,6 +230,12 @@ class SWFResponse(BaseResponse):
self.swf_backend.deprecate_domain(name)
return ""
def undeprecate_domain(self):
name = self._params["name"]
self._check_string(name)
self.swf_backend.undeprecate_domain(name)
return ""
def describe_domain(self):
name = self._params["name"]
self._check_string(name)
@ -278,6 +295,9 @@ class SWFResponse(BaseResponse):
def deprecate_activity_type(self):
return self._deprecate_type("activity")
def undeprecate_activity_type(self):
return self._undeprecate_type("activity")
def describe_activity_type(self):
return self._describe_type("activity")
@ -333,6 +353,9 @@ class SWFResponse(BaseResponse):
def deprecate_workflow_type(self):
return self._deprecate_type("workflow")
def undeprecate_workflow_type(self):
return self._undeprecate_type("workflow")
def describe_workflow_type(self):
return self._describe_type("workflow")

74
tests/test_swf/responses/test_activity_types.py
View File

@ -1,8 +1,11 @@
import boto
from boto.swf.exceptions import SWFResponseError
import boto3
from botocore.exceptions import ClientError
import sure # noqa
from moto import mock_swf_deprecated
from moto import mock_swf
# RegisterActivityType endpoint
@ -110,6 +113,77 @@ def test_deprecate_non_existent_activity_type():
).should.throw(SWFResponseError)
# DeprecateActivityType endpoint
@mock_swf
def test_undeprecate_activity_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.register_activity_type(
domain="test-domain", name="test-activity", version="v1.0"
)
client.deprecate_activity_type(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
)
client.undeprecate_activity_type(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
)
resp = client.describe_activity_type(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
)
resp["typeInfo"]["status"].should.equal("REGISTERED")
@mock_swf
def test_undeprecate_already_undeprecated_activity_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.register_activity_type(
domain="test-domain", name="test-activity", version="v1.0"
)
client.deprecate_activity_type(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
)
client.undeprecate_activity_type(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
)
client.undeprecate_activity_type.when.called_with(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
).should.throw(ClientError)
@mock_swf
def test_undeprecate_never_deprecated_activity_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.register_activity_type(
domain="test-domain", name="test-activity", version="v1.0"
)
client.undeprecate_activity_type.when.called_with(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
).should.throw(ClientError)
@mock_swf
def test_undeprecate_non_existent_activity_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.undeprecate_activity_type.when.called_with(
domain="test-domain", activityType={"name": "test-activity", "version": "v1.0"}
).should.throw(ClientError)
# DescribeActivityType endpoint
@mock_swf_deprecated
def test_describe_activity_type():

53
tests/test_swf/responses/test_domains.py
View File

@ -1,8 +1,11 @@
import boto
from boto.swf.exceptions import SWFResponseError
import boto3
from botocore.exceptions import ClientError
import sure # noqa
from moto import mock_swf_deprecated
from moto import mock_swf
# RegisterDomain endpoint
@ -94,6 +97,56 @@ def test_deprecate_non_existent_domain():
)
# UndeprecateDomain endpoint
@mock_swf
def test_undeprecate_domain():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.deprecate_domain(name="test-domain")
client.undeprecate_domain(name="test-domain")
resp = client.describe_domain(name="test-domain")
resp["domainInfo"]["status"].should.equal("REGISTERED")
@mock_swf
def test_undeprecate_already_undeprecated_domain():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.deprecate_domain(name="test-domain")
client.undeprecate_domain(name="test-domain")
client.undeprecate_domain.when.called_with(name="test-domain").should.throw(
ClientError
)
@mock_swf
def test_undeprecate_never_deprecated_domain():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.undeprecate_domain.when.called_with(name="test-domain").should.throw(
ClientError
)
@mock_swf
def test_undeprecate_non_existent_domain():
client = boto3.client("swf", region_name="us-east-1")
client.undeprecate_domain.when.called_with(name="non-existent").should.throw(
ClientError
)
# DescribeDomain endpoint
@mock_swf_deprecated
def test_describe_domain():

72
tests/test_swf/responses/test_workflow_types.py
View File

@ -5,6 +5,7 @@ import boto3
from moto import mock_swf_deprecated
from moto import mock_swf
from boto.swf.exceptions import SWFResponseError
from botocore.exceptions import ClientError
# RegisterWorkflowType endpoint
@ -112,6 +113,77 @@ def test_deprecate_non_existent_workflow_type():
).should.throw(SWFResponseError)
# UndeprecateWorkflowType endpoint
@mock_swf
def test_undeprecate_workflow_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.register_workflow_type(
domain="test-domain", name="test-workflow", version="v1.0"
)
client.deprecate_workflow_type(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
)
client.undeprecate_workflow_type(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
)
resp = client.describe_workflow_type(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
)
resp["typeInfo"]["status"].should.equal("REGISTERED")
@mock_swf
def test_undeprecate_already_undeprecated_workflow_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.register_workflow_type(
domain="test-domain", name="test-workflow", version="v1.0"
)
client.deprecate_workflow_type(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
)
client.undeprecate_workflow_type(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
)
client.undeprecate_workflow_type.when.called_with(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
).should.throw(ClientError)
@mock_swf
def test_undeprecate_never_deprecated_workflow_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.register_workflow_type(
domain="test-domain", name="test-workflow", version="v1.0"
)
client.undeprecate_workflow_type.when.called_with(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
).should.throw(ClientError)
@mock_swf
def test_undeprecate_non_existent_workflow_type():
client = boto3.client("swf", region_name="us-east-1")
client.register_domain(
name="test-domain", workflowExecutionRetentionPeriodInDays="60"
)
client.undeprecate_workflow_type.when.called_with(
domain="test-domain", workflowType={"name": "test-workflow", "version": "v1.0"}
).should.throw(ClientError)
# DescribeWorkflowType endpoint
@mock_swf_deprecated
def test_describe_workflow_type():