[Checkins] SVN: Sandbox/J1m/awsrecipes/src/zc/awsrecipes/ checkpoint
jim
cvs-admin at zope.org
Fri Jul 6 22:50:29 UTC 2012
Log message for revision 127278:
checkpoint
Changed:
U Sandbox/J1m/awsrecipes/src/zc/awsrecipes/__init__.py
U Sandbox/J1m/awsrecipes/src/zc/awsrecipes/ebs.test
U Sandbox/J1m/awsrecipes/src/zc/awsrecipes/tests.py
-=-
Modified: Sandbox/J1m/awsrecipes/src/zc/awsrecipes/__init__.py
===================================================================
--- Sandbox/J1m/awsrecipes/src/zc/awsrecipes/__init__.py 2012-07-06 20:11:04 UTC (rev 127277)
+++ Sandbox/J1m/awsrecipes/src/zc/awsrecipes/__init__.py 2012-07-06 22:50:25 UTC (rev 127278)
@@ -19,3 +19,41 @@
return ()
update = install
+
+user_data_start = '''#!/bin/sh
+hostname %s.aws.zope.net
+bcfg2
+'''
+
+class EC2:
+
+ def __init__(self, buildout, name, options):
+ self.options = options
+
+ def install(self):
+ options = self.options
+ conn = boto.ec2.connection.EC2Connection(region=options['region'])
+
+ [ami] = conn.get_all_images(
+ filters={'tag-key': 'Name', 'tag-value': 'default'})
+
+ user_data = user_data_start % options['name']
+
+ role = options.get('role')
+ if role:
+ user_data += '/opt/awshelpers/bin/set_role %s\n' % role
+
+ res = conn.run_instances(
+ ami.id,
+ security_groups = options.get(
+ 'security-groups', '').split() + ['default'],
+ placement = options['zone'],
+ user_data = user_data,
+ instance_type=options['type'],
+ )
+ conn.create_tags([instance.id for instance in res[0].instances],
+ dict(Name=self.options['name']))
+
+ return ()
+
+ update = install
Modified: Sandbox/J1m/awsrecipes/src/zc/awsrecipes/ebs.test
===================================================================
--- Sandbox/J1m/awsrecipes/src/zc/awsrecipes/ebs.test 2012-07-06 20:11:04 UTC (rev 127277)
+++ Sandbox/J1m/awsrecipes/src/zc/awsrecipes/ebs.test 2012-07-06 22:50:25 UTC (rev 127278)
@@ -19,7 +19,7 @@
>>> import zc.awsrecipes
>>> zc.awsrecipes.EBS({}, 'myvolume', dict(
- ... region = 'funky',
+ ... region = 'test_region',
... zone = 'us-east-1b',
... size = '50',
... name = 'data',
@@ -27,16 +27,11 @@
()
>>> import boto.ec2.connection
- >>> boto.ec2.connection.EC2Connection.assert_called_with(region='funky')
- >>> conn = boto.ec2.connection.EC2Connection.return_value
+ >>> conn = boto.ec2.connection.EC2Connection(region='test_region')
- >>> conn.create_volume.assert_called_with(50, 'us-east-1b')
-
Now, if we look throug our ebs volumes, we should see 2 with a name of
data:
- >>> import boto.ec2.connection
- >>> conn = conn = boto.ec2.connection.EC2Connection()
>>> for v in conn.get_all_volumes():
... if v.tags['Name'] == 'data':
... print v.size
Modified: Sandbox/J1m/awsrecipes/src/zc/awsrecipes/tests.py
===================================================================
--- Sandbox/J1m/awsrecipes/src/zc/awsrecipes/tests.py 2012-07-06 20:11:04 UTC (rev 127277)
+++ Sandbox/J1m/awsrecipes/src/zc/awsrecipes/tests.py 2012-07-06 22:50:25 UTC (rev 127278)
@@ -22,43 +22,90 @@
return lambda f: side_effect(m, f)
m.side_effect = f
-def ebssetup(test):
- volumes = []
- EC2Connection = setupstack.context_manager(
- test, mock.patch('boto.ec2.connection.EC2Connection'))
+def assert_(cond, mess='assertion failed'):
+ if not cond:
+ raise AssertionError(mess)
- class Volume:
- def __init__(self, id, size, zone):
- self.id = id
- self.size = size
- self.zone = zone
- self.tags = {}
+class Resource:
+ def __init__(self, id, tags=None, attrs=None, **kw):
+ self.id = id
+ self.tags = tags or {}
+ self.attrs = attrs or {}
+ self.__dict__.update(kw)
- @side_effect(EC2Connection.return_value.create_volume)
- def create_volume(size, zone):
- volume = Volume(str(len(volumes)), size, zone)
- volumes.append(volume)
+def filter(data, filters):
+ assert_(filters['tag-key'] == 'Name')
+ return [
+ r for r in data
+ if r.tags['Name'] == filters['tag-value']
+ ]
+
+class Connection:
+
+ def __init__(self):
+ self.volumes = []
+ self.instances = []
+ self.resources = {}
+ self.groups = dict(
+ default=Resource('gr1'),
+ sg1=Resource('gr2'),
+ )
+
+ def create_volume(self, size, zone):
+ assert_(zone == 'us-east-1b')
+ volume = Resource('vol%s' % len(self.volumes), size=size, zone=zone)
+ self.volumes.append(volume)
+ self.resources[volume.id] = volume
return volume
- @side_effect(EC2Connection.return_value.create_tags)
- def create_tags(ids, tags):
- for v in volumes:
- if v.id in ids:
- v.tags.update(tags)
+ def create_tags(self, ids, tags):
+ for id in ids:
+ self.resources[id].tags.update(tags)
- @side_effect(EC2Connection.return_value.get_all_volumes)
- def get_all_volumes(ids=None):
+ def get_all_volumes(self, ids=None):
if ids is None:
- return list(volumes)
+ return list(self.volumes)
else:
- return [v for v in volumes if v.id in ids]
+ return [v for v in self.volumes if v.id in ids]
+ def get_all_images(self, filters):
+ return [Resource('ami-42')]
+ def get_all_instances(self, filters):
+ return [Resource('', instances=filter(self.instances, filters))]
+
+ def run_instances(
+ self, ami_id, security_groups, placement, user_data, instance_type):
+ id = 'inst%s' % len(self.instances)
+ instance = Resource(
+ id,
+ image_id=ami_id,
+ placement=placement,
+ attrs=dict(
+ userData=user_data.encode('base64'),
+ groupSet=[self.groups[name] for name in security_groups],
+ ),
+ )
+ self.instances.append(instance)
+ self.resources[instance.id] = instance
+ return [Resource('', instances=[instance])]
+
+def setup(test):
+ connections = dict(test_region=Connection())
+
+ @side_effect(
+ setupstack.context_manager(
+ test, mock.patch('boto.ec2.connection.EC2Connection')))
+ def connect(region):
+ return connections[region]
+
+
+
def test_suite():
return unittest.TestSuite((
doctest.DocFileSuite(
- 'ebs.test',
- setUp=ebssetup, tearDown=setupstack.tearDown),
+ 'ebs.test', 'ec2.test',
+ setUp=setup, tearDown=setupstack.tearDown),
))
if __name__ == '__main__':
More information about the checkins
mailing list