| OLD | NEW |
|---|
| (Empty) |
| 1 # Copyright (c) 2006-2010 Mitch Garnaat http://garnaat.org/ | |
| 2 # Copyright (c) 2009, Eucalyptus Systems, Inc. | |
| 3 # All rights reserved. | |
| 4 # | |
| 5 # Permission is hereby granted, free of charge, to any person obtaining a | |
| 6 # copy of this software and associated documentation files (the | |
| 7 # "Software"), to deal in the Software without restriction, including | |
| 8 # without limitation the rights to use, copy, modify, merge, publish, dis- | |
| 9 # tribute, sublicense, and/or sell copies of the Software, and to permit | |
| 10 # persons to whom the Software is furnished to do so, subject to the fol- | |
| 11 # lowing conditions: | |
| 12 # | |
| 13 # The above copyright notice and this permission notice shall be included | |
| 14 # in all copies or substantial portions of the Software. | |
| 15 # | |
| 16 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS | |
| 17 # OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABIL- | |
| 18 # ITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT | |
| 19 # SHALL THE AUTHOR BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, | |
| 20 # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| 21 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS | |
| 22 # IN THE SOFTWARE. | |
| 23 | |
| 24 """ | |
| 25 Some unit tests for the EC2Connection | |
| 26 """ | |
| 27 | |
| 28 import unittest | |
| 29 import time | |
| 30 from boto.ec2.connection import EC2Connection | |
| 31 import telnetlib | |
| 32 import socket | |
| 33 | |
| 34 class EC2ConnectionTest (unittest.TestCase): | |
| 35 | |
| 36 def test_1_basic(self): | |
| 37 # this is my user_id, if you want to run these tests you should | |
| 38 # replace this with yours or they won't work | |
| 39 user_id = '963068290131' | |
| 40 print '--- running EC2Connection tests ---' | |
| 41 c = EC2Connection() | |
| 42 # get list of private AMI's | |
| 43 rs = c.get_all_images(owners=[user_id]) | |
| 44 assert len(rs) > 0 | |
| 45 # now pick the first one | |
| 46 image = rs[0] | |
| 47 # temporarily make this image runnable by everyone | |
| 48 status = image.set_launch_permissions(group_names=['all']) | |
| 49 assert status | |
| 50 d = image.get_launch_permissions() | |
| 51 assert d.has_key('groups') | |
| 52 assert len(d['groups']) > 0 | |
| 53 # now remove that permission | |
| 54 status = image.remove_launch_permissions(group_names=['all']) | |
| 55 assert status | |
| 56 time.sleep(10) | |
| 57 d = image.get_launch_permissions() | |
| 58 assert not d.has_key('groups') | |
| 59 | |
| 60 # create 2 new security groups | |
| 61 group1_name = 'test-%d' % int(time.time()) | |
| 62 group_desc = 'This is a security group created during unit testing' | |
| 63 group1 = c.create_security_group(group1_name, group_desc) | |
| 64 time.sleep(2) | |
| 65 group2_name = 'test-%d' % int(time.time()) | |
| 66 group_desc = 'This is a security group created during unit testing' | |
| 67 group2 = c.create_security_group(group2_name, group_desc) | |
| 68 # now get a listing of all security groups and look for our new one | |
| 69 rs = c.get_all_security_groups() | |
| 70 found = False | |
| 71 for g in rs: | |
| 72 if g.name == group1_name: | |
| 73 found = True | |
| 74 assert found | |
| 75 # now pass arg to filter results to only our new group | |
| 76 rs = c.get_all_security_groups([group1_name]) | |
| 77 assert len(rs) == 1 | |
| 78 # try some group to group authorizations/revocations | |
| 79 # first try the old style | |
| 80 status = c.authorize_security_group(group1.name, group2.name, group2.own
er_id) | |
| 81 assert status | |
| 82 status = c.revoke_security_group(group1.name, group2.name, group2.owner_
id) | |
| 83 assert status | |
| 84 # now try specifying a specific port | |
| 85 status = c.authorize_security_group(group1.name, group2.name, group2.own
er_id, | |
| 86 'tcp', 22, 22) | |
| 87 assert status | |
| 88 status = c.revoke_security_group(group1.name, group2.name, group2.owner_
id, | |
| 89 'tcp', 22, 22) | |
| 90 assert status | |
| 91 | |
| 92 # now delete the second security group | |
| 93 status = c.delete_security_group(group2_name) | |
| 94 # now make sure it's really gone | |
| 95 rs = c.get_all_security_groups() | |
| 96 found = False | |
| 97 for g in rs: | |
| 98 if g.name == group2_name: | |
| 99 found = True | |
| 100 assert not found | |
| 101 | |
| 102 group = group1 | |
| 103 | |
| 104 # now try to launch apache image with our new security group | |
| 105 rs = c.get_all_images() | |
| 106 img_loc = 'ec2-public-images/fedora-core4-apache.manifest.xml' | |
| 107 for image in rs: | |
| 108 if image.location == img_loc: | |
| 109 break | |
| 110 reservation = image.run(security_groups=[group.name]) | |
| 111 instance = reservation.instances[0] | |
| 112 while instance.state != 'running': | |
| 113 print '\tinstance is %s' % instance.state | |
| 114 time.sleep(30) | |
| 115 instance.update() | |
| 116 # instance in now running, try to telnet to port 80 | |
| 117 t = telnetlib.Telnet() | |
| 118 try: | |
| 119 t.open(instance.dns_name, 80) | |
| 120 except socket.error: | |
| 121 pass | |
| 122 # now open up port 80 and try again, it should work | |
| 123 group.authorize('tcp', 80, 80, '0.0.0.0/0') | |
| 124 t.open(instance.dns_name, 80) | |
| 125 t.close() | |
| 126 # now revoke authorization and try again | |
| 127 group.revoke('tcp', 80, 80, '0.0.0.0/0') | |
| 128 try: | |
| 129 t.open(instance.dns_name, 80) | |
| 130 except socket.error: | |
| 131 pass | |
| 132 # now kill the instance and delete the security group | |
| 133 instance.terminate() | |
| 134 # unfortunately, I can't delete the sg within this script | |
| 135 #sg.delete() | |
| 136 | |
| 137 # create a new key pair | |
| 138 key_name = 'test-%d' % int(time.time()) | |
| 139 status = c.create_key_pair(key_name) | |
| 140 assert status | |
| 141 # now get a listing of all key pairs and look for our new one | |
| 142 rs = c.get_all_key_pairs() | |
| 143 found = False | |
| 144 for k in rs: | |
| 145 if k.name == key_name: | |
| 146 found = True | |
| 147 assert found | |
| 148 # now pass arg to filter results to only our new key pair | |
| 149 rs = c.get_all_key_pairs([key_name]) | |
| 150 assert len(rs) == 1 | |
| 151 key_pair = rs[0] | |
| 152 # now delete the key pair | |
| 153 status = c.delete_key_pair(key_name) | |
| 154 # now make sure it's really gone | |
| 155 rs = c.get_all_key_pairs() | |
| 156 found = False | |
| 157 for k in rs: | |
| 158 if k.name == key_name: | |
| 159 found = True | |
| 160 assert not found | |
| 161 | |
| 162 # short test around Paid AMI capability | |
| 163 demo_paid_ami_id = 'ami-bd9d78d4' | |
| 164 demo_paid_ami_product_code = 'A79EC0DB' | |
| 165 l = c.get_all_images([demo_paid_ami_id]) | |
| 166 assert len(l) == 1 | |
| 167 assert len(l[0].product_codes) == 1 | |
| 168 assert l[0].product_codes[0] == demo_paid_ami_product_code | |
| 169 | |
| 170 print '--- tests completed ---' | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 10199002:
Upgrade gsutil to 3.4 (Closed)
Base URL: https://dart.googlecode.com/svn/branches/bleeding_edge/dart