Commit 79c669e1 by Simon János

basic model implementation: resources and resource models

parent fcde0eb5
import uuid
from enum import Enum
from voluptuous import Schema, Invalid, MultipleInvalid
class Resource(object):
DEFAULT_ID_LENGTH = 8
def __init__(self, *args, **kwargs):
if len(args):
resource = args[0]
if isinstance(resource, Resource):
self.__id = resource.id
self.__type = Resource.Type[resource.type]
else:
self.__attributes_from_dict(resource)
else:
self.__attributes_from_dict(kwargs)
def __attributes_from_dict(self, attributes):
schema = Schema({
'id': str,
'type': Resource.Type.validate
})
try:
schema(attributes)
self.__id = attributes.get('id', Resource._random_id())
self.__type = Resource.Type[attributes.get('type', 'null')]
except MultipleInvalid as invalid:
raise InvalidResourceException(invalid)
@property
def id(self):
return self.__id
@property
def type(self):
return self.__type.name
@staticmethod
def _random_id():
return uuid.uuid4().hex[:Resource.DEFAULT_ID_LENGTH]
def __eq__(self, other):
return isinstance(other, Resource) and \
self.id == other.id and \
self.type == other.type
def __repr__(self):
return "%s(%s)" % (self.__class__.__name__, self.id)
def __hash__(self):
return hash(self.__repr__())
class Type(Enum):
null = 0
instance = 1
@classmethod
def validate(cls, value):
try:
cls[value]
except KeyError:
raise Invalid('The given resource type (%s) is not valid' % value)
return True
class Instance(Resource):
def __init__(self, *args, **kwargs):
kwargs['type'] = Resource.Type.instance.name
super(Instance, self).__init__(*args, **kwargs)
class ResourceGroup(object):
def __init__(self, *args):
self.__resources = set()
if len(args):
for argument in args:
try:
self.__iadd__(argument)
except InvalidResourceException:
for resource in argument:
self.__iadd__(resource)
@property
def resources(self):
return list(self.__resources)
def __add__(self, other):
if isinstance(other, dict):
return ResourceGroup(Resource(other), *self.resources)
if isinstance(other, Resource):
return ResourceGroup(other, *self.resources)
if isinstance(other, ResourceGroup):
resources = other.resources + self.resources
return ResourceGroup(*resources)
raise InvalidResourceException()
def __iadd__(self, other):
if isinstance(other, dict):
self.__resources.add(Resource(other))
return self
if isinstance(other, Resource):
self.__resources.add(other)
return self
if isinstance(other, ResourceGroup):
self.__resources.union(other.resources)
return self
raise InvalidResourceException()
def __iter__(self):
return self.__resources.__iter__()
def __eq__(self, other):
return self.__resources == set(other.resources)
class InvalidResourceException(Exception):
pass
from unittest.case import TestCase
from voluptuous import Invalid
from orchestrator.model.resources import Resource, ResourceGroup, Instance, InvalidResourceException
class ResourceTest(TestCase):
def test_resource_without_args(self):
# when
resource = Resource()
# then
self.assertIsInstance(resource.id, str)
self.assertEqual(Resource.DEFAULT_ID_LENGTH, len(resource.id))
self.assertEqual('null', resource.type)
def test_resource_from_args(self):
# given
expected_id = 'dummy_resource'
expected_type = 'null'
# when
result = Resource(id=expected_id, type=expected_type)
# then
self.assertEqual(expected_id, result.id)
self.assertEqual(expected_type, result.type)
def test_resource_from_dict(self):
# given
resource_dict = dict(id='dummy_resource', type='null')
# when
result = Resource(resource_dict)
# then
self.assertEqual(resource_dict['id'], result.id)
self.assertEqual(resource_dict['type'], result.type)
def test_resource_from_invalid_args(self):
with self.assertRaises(InvalidResourceException):
Resource(id='dummy_id', invalid_attribute=False)
def test_resource_from_invalid_dict(self):
# given
invalid_attributes = dict(id='dummy_id', invalid_attribute=False)
# when/then
with self.assertRaises(InvalidResourceException):
Resource(invalid_attributes)
def test_resource_type_validate(self):
self.assertTrue(Resource.Type.validate('instance'))
def test_resource_type_invalid(self):
with self.assertRaises(Invalid):
Resource.Type.validate('nonexisting_resource_type')
def test_resource_instance(self):
# when
instance = Instance()
# then
self.assertIsInstance(instance.id, str)
self.assertEqual(Resource.DEFAULT_ID_LENGTH, len(instance.id))
self.assertEqual(Resource.Type.instance.name, instance.type)
def test_resource_equivalence(self):
# given
resource1 = Resource(id='some_resource', type='null')
resource2 = Resource(id='some_resource', type='null')
resource3 = Resource(id='some_other_resource', type='null')
resource4 = Resource(id='some_resource', type='instance')
# then
self.assertEqual(resource1, resource1)
self.assertEqual(resource1, resource2)
self.assertNotEqual(resource1, resource3)
self.assertNotEqual(resource1, resource4)
class ResourceGroupTest(TestCase):
def test_resource_group_without_args(self):
# when
empty_group = ResourceGroup()
# then
self.assertEqual([], empty_group.resources)
def test_resource_group_from_resources(self):
# given
resource1 = Resource(id='some_resource')
resource2 = Resource(id='some_other_resource')
# then
self.assertEqual([resource1], ResourceGroup(resource1).resources)
self.assertEqual([resource1], ResourceGroup([resource1]).resources)
self.assertEqual([resource1], ResourceGroup([resource1, resource1]).resources)
for resources in [[resource1, resource2], [resource2, resource1]]:
self.assertEqual(2, len(ResourceGroup(*resources).resources))
self.assertIn(resource1, ResourceGroup(*resources).resources)
self.assertIn(resource2, ResourceGroup(*resources).resources)
def test_resource_group_from_dicts(self):
# given
dict1 = dict(id='some_resource')
dict2 = dict(id='some_other_resource')
expected1 = Resource(dict1)
expected2 = Resource(dict2)
# then
self.assertEqual([expected1], ResourceGroup(dict1).resources)
self.assertEqual([expected1], ResourceGroup([dict1]).resources)
self.assertEqual([expected1], ResourceGroup([dict1, dict1]).resources)
for resources in [[expected1, expected2], [dict2, dict1]]:
self.assertEqual(2, len(ResourceGroup(resources).resources))
self.assertIn(expected1, ResourceGroup(resources).resources)
self.assertIn(expected2, ResourceGroup(resources).resources)
def test_resource_group_addition(self):
# given
resource1 = Resource()
resource2 = Resource()
group1 = ResourceGroup(resource1)
group2 = ResourceGroup(resource2)
# when
result_group = group1 + group2
# then
self.assertEqual(2, len(result_group.resources))
self.assertIn(resource1, result_group.resources)
self.assertIn(resource2, result_group.resources)
def test_resource_group_iteration(self):
# when
group = ResourceGroup(Resource(), Resource(), Resource())
# then
for index, resource in enumerate(group):
self.assertEqual(group.resources[index], resource)
def test_resource_group_equivalence(self):
# given
resource1 = Resource(id='some_resource', type='null')
resource2 = Resource(id='some_resource', type='null')
resource3 = Resource(id='some_other_resource', type='null')
# then
self.assertEqual(ResourceGroup(), ResourceGroup())
self.assertEqual(ResourceGroup(resource1), ResourceGroup(resource1))
self.assertEqual(ResourceGroup(resource1), ResourceGroup(resource2))
self.assertNotEqual(ResourceGroup(resource1), ResourceGroup(resource3))
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment