← Back to branch summary

~ntt-pf-lab/nova/monkey_patch_notification

« back to all changes in this revision

Viewing changes to vendor/boto/boto/tests/test_ec2connection.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/boto/boto/tests/test_ec2connection.py
 
1
#!/usr/bin/env python
 
2
 
 
3
# Copyright (c) 2006,2007 Mitch Garnaat http://garnaat.org/
 
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
import os
 
31
from boto.ec2.connection import EC2Connection
 
32
import telnetlib
 
33
import socket
 
34
 
 
35
class EC2ConnectionTest (unittest.TestCase):
 
36
 
 
37
    def test_1_basic(self):
 
38
        # this is my user_id, if you want to run these tests you should
 
39
        # replace this with yours or they won't work
 
40
        user_id = '963068290131'
 
41
        print '--- running EC2Connection tests ---'
 
42
        c = EC2Connection()
 
43
        # get list of private AMI's
 
44
        rs = c.get_all_images(owners=[user_id])
 
45
        assert len(rs) > 0
 
46
        # now pick the first one
 
47
        image = rs[0]
 
48
        # temporarily make this image runnable by everyone
 
49
        status = image.set_launch_permissions(group_names=['all'])
 
50
        assert status
 
51
        d = image.get_launch_permissions()
 
52
        assert d.has_key('groups')
 
53
        assert len(d['groups']) > 0
 
54
        # now remove that permission
 
55
        status = image.remove_launch_permissions(group_names=['all'])
 
56
        assert status
 
57
        d = image.get_launch_permissions()
 
58
        assert not d.has_key('groups')
 
59
        
 
60
        # create a new security group
 
61
        group_name = 'test-%d' % int(time.time())
 
62
        group_desc = 'This is a security group created during unit testing'
 
63
        group = c.create_security_group(group_name, group_desc)
 
64
        # now get a listing of all security groups and look for our new one
 
65
        rs = c.get_all_security_groups()
 
66
        found = False
 
67
        for g in rs:
 
68
            if g.name == group_name:
 
69
                found = True
 
70
        assert found
 
71
        # now pass arg to filter results to only our new group
 
72
        rs = c.get_all_security_groups([group_name])
 
73
        assert len(rs) == 1
 
74
        group = rs[0]
 
75
        #
 
76
        # now delete the security group
 
77
        status = c.delete_security_group(group_name)
 
78
        # now make sure it's really gone
 
79
        rs = c.get_all_security_groups()
 
80
        found = False
 
81
        for g in rs:
 
82
            if g.name == group_name:
 
83
                found = True
 
84
        assert not found
 
85
        # now create it again for use with the instance test
 
86
        group = c.create_security_group(group_name, group_desc)
 
87
        
 
88
        # now try to launch apache image with our new security group
 
89
        rs = c.get_all_images()
 
90
        img_loc = 'ec2-public-images/fedora-core4-apache.manifest.xml'
 
91
        for image in rs:
 
92
            if image.location == img_loc:
 
93
                break
 
94
        reservation = image.run(security_groups=[group.name])
 
95
        instance = reservation.instances[0]
 
96
        while instance.state != 'running':
 
97
            print '\tinstance is %s' % instance.state
 
98
            time.sleep(30)
 
99
            instance.update()
 
100
        # instance in now running, try to telnet to port 80
 
101
        t = telnetlib.Telnet()
 
102
        try:
 
103
            t.open(instance.dns_name, 80)
 
104
        except socket.error:
 
105
            pass
 
106
        # now open up port 80 and try again, it should work
 
107
        group.authorize('tcp', 80, 80, '0.0.0.0/0')
 
108
        t.open(instance.dns_name, 80)
 
109
        t.close()
 
110
        # now revoke authorization and try again
 
111
        group.revoke('tcp', 80, 80, '0.0.0.0/0')
 
112
        try:
 
113
            t.open(instance.dns_name, 80)
 
114
        except socket.error:
 
115
            pass
 
116
        # now kill the instance and delete the security group
 
117
        instance.stop()
 
118
        # unfortunately, I can't delete the sg within this script
 
119
        #sg.delete()
 
120
        
 
121
        # create a new key pair
 
122
        key_name = 'test-%d' % int(time.time())
 
123
        status = c.create_key_pair(key_name)
 
124
        assert status
 
125
        # now get a listing of all key pairs and look for our new one
 
126
        rs = c.get_all_key_pairs()
 
127
        found = False
 
128
        for k in rs:
 
129
            if k.name == key_name:
 
130
                found = True
 
131
        assert found
 
132
        # now pass arg to filter results to only our new key pair
 
133
        rs = c.get_all_key_pairs([key_name])
 
134
        assert len(rs) == 1
 
135
        key_pair = rs[0]
 
136
        # now delete the key pair
 
137
        status = c.delete_key_pair(key_name)
 
138
        # now make sure it's really gone
 
139
        rs = c.get_all_key_pairs()
 
140
        found = False
 
141
        for k in rs:
 
142
            if k.name == key_name:
 
143
                found = True
 
144
        assert not found
 
145
 
 
146
        # short test around Paid AMI capability
 
147
        demo_paid_ami_id = 'ami-bd9d78d4'
 
148
        demo_paid_ami_product_code = 'A79EC0DB'
 
149
        l = c.get_all_images([demo_paid_ami_id])
 
150
        assert len(l) == 1
 
151
        assert len(l[0].product_codes) == 1
 
152
        assert l[0].product_codes[0] == demo_paid_ami_product_code
 
153
        
 
154
        print '--- tests completed ---'