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

Abstract away SWF *Type models logic into a GenericType class

Browse Source
This commit is contained in:
Jean-Baptiste Barth 2015-10-01 21:25:25 +02:00
parent 9483355584
commit 5c02fcd94b
3 changed files with 72 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
moto/swf/models.py
View File

@ -68,7 +68,7 @@ class Domain(object):
return _all
class ActivityType(object):
class GenericType(object):
def __init__(self, name, version, **kwargs):
self.name = name
self.version = version
@ -78,17 +78,13 @@ class ActivityType(object):
for key, value in kwargs.iteritems():
self.__setattr__(key, value)
def __repr__(self):
return "ActivityType(name: %(name)s, version: %(version)s)" % self.__dict__
@property
def kind(self):
raise NotImplementedError()
@property
def _configuration_keys(self):
return [
"defaultTaskHeartbeatTimeout",
"defaultTaskScheduleToCloseTimeout",
"defaultTaskScheduleToStartTimeout",
"defaultTaskStartToCloseTimeout",
]
raise NotImplementedError()
def to_short_dict(self):
return {
@ -98,7 +94,7 @@ class ActivityType(object):
def to_medium_dict(self):
hsh = {
"activityType": self.to_short_dict(),
"{}Type".format(self.kind): self.to_short_dict(),
"creationDate": 1420066800,
"status": self.status,
}
@ -124,15 +120,25 @@ class ActivityType(object):
hsh["configuration"][key] = getattr(self, attr)
return hsh
class ActivityType(GenericType):
def __repr__(self):
return "ActivityType(name: %(name)s, version: %(version)s)" % self.__dict__
class WorkflowType(object):
def __init__(self, name, version, **kwargs):
self.name = name
self.version = version
self.status = "REGISTERED"
for key, value in kwargs.iteritems():
self.__setattr__(key, value)
@property
def _configuration_keys(self):
return [
"defaultTaskHeartbeatTimeout",
"defaultTaskScheduleToCloseTimeout",
"defaultTaskScheduleToStartTimeout",
"defaultTaskStartToCloseTimeout",
]
@property
def kind(self):
return "activity"
class WorkflowType(GenericType):
def __repr__(self):
return "WorkflowType(name: %(name)s, version: %(version)s)" % self.__dict__
@ -144,39 +150,9 @@ class WorkflowType(object):
"defaultTaskStartToCloseTimeout",
]
def to_short_dict(self):
return {
"name": self.name,
"version": self.version,
}
def to_medium_dict(self):
hsh = {
"workflowType": self.to_short_dict(),
"creationDate": 1420066800,
"status": self.status,
}
if self.status == "DEPRECATED":
hsh["deprecationDate"] = 1422745200
if hasattr(self, "description"):
hsh["description"] = self.description
return hsh
def to_full_dict(self):
hsh = {
"typeInfo": self.to_medium_dict(),
"configuration": {}
}
if hasattr(self, "task_list"):
hsh["configuration"]["defaultTaskList"] = {"name": self.task_list}
for key in self._configuration_keys:
attr = camelcase_to_underscores(key)
if not hasattr(self, attr):
continue
if getattr(self, attr) is None:
continue
hsh["configuration"][key] = getattr(self, attr)
return hsh
@property
def kind(self):
return "workflow"
class SWFBackend(BaseBackend):

31
tests/test_swf/test_activity_types.py
View File

@ -12,37 +12,6 @@ from moto.swf.exceptions import (
)
# Models
def test_short_dict_representation():
_type = ActivityType("test-activity", "v1.0")
_type.to_short_dict().should.equal({"name": "test-activity", "version": "v1.0"})
def test_medium_dict_representation():
_type = ActivityType("test-activity", "v1.0")
_type.to_medium_dict()["activityType"].should.equal(_type.to_short_dict())
_type.to_medium_dict()["status"].should.equal("REGISTERED")
_type.to_medium_dict().should.contain("creationDate")
_type.to_medium_dict().should_not.contain("deprecationDate")
_type.to_medium_dict().should_not.contain("description")
_type.description = "foo bar"
_type.to_medium_dict()["description"].should.equal("foo bar")
_type.status = "DEPRECATED"
_type.to_medium_dict().should.contain("deprecationDate")
def test_full_dict_representation():
_type = ActivityType("test-activity", "v1.0")
_type.to_full_dict()["typeInfo"].should.equal(_type.to_medium_dict())
_type.to_full_dict()["configuration"].should.equal({})
_type.task_list = "foo"
_type.to_full_dict()["configuration"]["defaultTaskList"].should.equal({"name":"foo"})
_type.default_task_heartbeat_timeout = "60"
_type.to_full_dict()["configuration"]["defaultTaskHeartbeatTimeout"].should.equal("60")
# RegisterActivityType endpoint
@mock_swf
def test_register_activity_type():

46
tests/test_swf/test_models.py Normal file
View File

@ -0,0 +1,46 @@
from sure import expect
from moto.swf.models import GenericType
class FooType(GenericType):
@property
def kind(self):
return "foo"
@property
def _configuration_keys(self):
return ["justAnExampleTimeout"]
def test_short_dict_representation():
_type = FooType("test-foo", "v1.0")
_type.to_short_dict().should.equal({"name": "test-foo", "version": "v1.0"})
def test_medium_dict_representation():
_type = FooType("test-foo", "v1.0")
_type.to_medium_dict()["fooType"].should.equal(_type.to_short_dict())
_type.to_medium_dict()["status"].should.equal("REGISTERED")
_type.to_medium_dict().should.contain("creationDate")
_type.to_medium_dict().should_not.contain("deprecationDate")
_type.to_medium_dict().should_not.contain("description")
_type.description = "foo bar"
_type.to_medium_dict()["description"].should.equal("foo bar")
_type.status = "DEPRECATED"
_type.to_medium_dict().should.contain("deprecationDate")
def test_full_dict_representation():
_type = FooType("test-foo", "v1.0")
_type.to_full_dict()["typeInfo"].should.equal(_type.to_medium_dict())
_type.to_full_dict()["configuration"].should.equal({})
_type.task_list = "foo"
_type.to_full_dict()["configuration"]["defaultTaskList"].should.equal({"name":"foo"})
_type.just_an_example_timeout = "60"
_type.to_full_dict()["configuration"]["justAnExampleTimeout"].should.equal("60")
_type.non_whitelisted_property = "34"
_type.to_full_dict()["configuration"].keys().should.equal(["defaultTaskList", "justAnExampleTimeout"])