← Back to branch summary

~cbehrens/nova/lp844160-build-works-with-zones

« back to all changes in this revision

Viewing changes to smoketests/public_network_smoketests.py

  • Committer: NTT PF Lab.
  • Date: 2010-12-24 11:38:49 UTC
  • mto: This revision was merged to the branch mainline in revision 564.
  • Revision ID: openstack@lab.ntt.co.jp-20101224113849-z9nemzmki17bxnvw
Support IPv6

Show diffs side-by-side

added added

removed removed

Lines of Context:
smoketests/public_network_smoketests.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2010 United States Government as represented by the
 
4
# Administrator of the National Aeronautics and Space Administration.
 
5
# All Rights Reserved.
 
6
#
 
7
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
8
#    not use this file except in compliance with the License. You may obtain
 
9
#    a copy of the License at
 
10
#
 
11
#         http://www.apache.org/licenses/LICENSE-2.0
 
12
#
 
13
#    Unless required by applicable law or agreed to in writing, software
 
14
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
15
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
16
#    License for the specific language governing permissions and limitations
 
17
#    under the License.
 
18
 
 
19
import commands
 
20
import os
 
21
import random
 
22
import socket
 
23
import sys
 
24
import time
 
25
import unittest
 
26
 
 
27
from smoketests import flags
 
28
from smoketests import base
 
29
from smoketests import user_smoketests
 
30
 
 
31
#Note that this test should run from
 
32
#public network (outside of private network segments)
 
33
#Please set EC2_URL correctly
 
34
#You should use admin account in this test
 
35
 
 
36
FLAGS = flags.FLAGS
 
37
 
 
38
TEST_PREFIX = 'test%s' % int(random.random() * 1000000)
 
39
TEST_BUCKET = '%s_bucket' % TEST_PREFIX
 
40
TEST_KEY = '%s_key' % TEST_PREFIX
 
41
TEST_KEY2 = '%s_key2' % TEST_PREFIX
 
42
TEST_DATA = {}
 
43
 
 
44
 
 
45
class InstanceTestsFromPublic(user_smoketests.UserSmokeTestCase):
 
46
    def test_001_can_create_keypair(self):
 
47
        key = self.create_key_pair(self.conn, TEST_KEY)
 
48
        self.assertEqual(key.name, TEST_KEY)
 
49
 
 
50
    def test_002_security_group(self):
 
51
        security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd")
 
52
                                      for x in range(random.randint(4, 8)))
 
53
        group = self.conn.create_security_group(security_group_name,
 
54
                                               'test group')
 
55
        group.connection = self.conn
 
56
        group.authorize('tcp', 22, 22, '0.0.0.0/0')
 
57
        if FLAGS.use_ipv6:
 
58
            group.authorize('tcp', 22, 22, '::/0')
 
59
 
 
60
        reservation = self.conn.run_instances(FLAGS.test_image,
 
61
                                        key_name=TEST_KEY,
 
62
                                        security_groups=[security_group_name],
 
63
                                        instance_type='m1.tiny')
 
64
        self.data['security_group_name'] = security_group_name
 
65
        self.data['group'] = group
 
66
        self.data['instance_id'] = reservation.instances[0].id
 
67
 
 
68
    def test_003_instance_with_group_runs_within_60_seconds(self):
 
69
        reservations = self.conn.get_all_instances([self.data['instance_id']])
 
70
        instance = reservations[0].instances[0]
 
71
        # allow 60 seconds to exit pending with IP
 
72
        for x in xrange(60):
 
73
            instance.update()
 
74
            if instance.state == u'running':
 
75
                break
 
76
            time.sleep(1)
 
77
        else:
 
78
            self.fail('instance failed to start')
 
79
        ip = reservations[0].instances[0].private_dns_name
 
80
        self.failIf(ip == '0.0.0.0')
 
81
        self.data['private_ip'] = ip
 
82
        if FLAGS.use_ipv6:
 
83
            ipv6 = reservations[0].instances[0].dns_name_v6
 
84
            self.failIf(ipv6 is None)
 
85
            self.data['ip_v6'] = ipv6
 
86
 
 
87
    def test_004_can_ssh_to_ipv6(self):
 
88
        if FLAGS.use_ipv6:
 
89
            for x in xrange(20):
 
90
                try:
 
91
                    conn = self.connect_ssh(
 
92
                                        self.data['ip_v6'], TEST_KEY)
 
93
                    conn.close()
 
94
                except Exception as ex:
 
95
                    print ex
 
96
                    time.sleep(1)
 
97
                else:
 
98
                    break
 
99
            else:
 
100
                self.fail('could not ssh to instance')
 
101
 
 
102
    def test_012_can_create_instance_with_keypair(self):
 
103
        if 'instance_id' in self.data:
 
104
            self.conn.terminate_instances([self.data['instance_id']])
 
105
        reservation = self.conn.run_instances(FLAGS.test_image,
 
106
                                              key_name=TEST_KEY,
 
107
                                              instance_type='m1.tiny')
 
108
        self.assertEqual(len(reservation.instances), 1)
 
109
        self.data['instance_id'] = reservation.instances[0].id
 
110
 
 
111
    def test_013_instance_runs_within_60_seconds(self):
 
112
        reservations = self.conn.get_all_instances([self.data['instance_id']])
 
113
        instance = reservations[0].instances[0]
 
114
        # allow 60 seconds to exit pending with IP
 
115
        for x in xrange(60):
 
116
            instance.update()
 
117
            if instance.state == u'running':
 
118
                break
 
119
            time.sleep(1)
 
120
        else:
 
121
            self.fail('instance failed to start')
 
122
        ip = reservations[0].instances[0].private_dns_name
 
123
        self.failIf(ip == '0.0.0.0')
 
124
        self.data['private_ip'] = ip
 
125
        if FLAGS.use_ipv6:
 
126
            ipv6 = reservations[0].instances[0].dns_name_v6
 
127
            self.failIf(ipv6 is None)
 
128
            self.data['ip_v6'] = ipv6
 
129
 
 
130
    def test_014_can_not_ping_private_ip(self):
 
131
        for x in xrange(4):
 
132
            # ping waits for 1 second
 
133
            status, output = commands.getstatusoutput(
 
134
                'ping -c1 %s' % self.data['private_ip'])
 
135
            if status == 0:
 
136
                self.fail('can ping private ip from public network')
 
137
            if FLAGS.use_ipv6:
 
138
                status, output = commands.getstatusoutput(
 
139
                    'ping6 -c1 %s' % self.data['ip_v6'])
 
140
                if status == 0:
 
141
                    self.fail('can ping ipv6 from public network')
 
142
        else:
 
143
            pass
 
144
 
 
145
    def test_015_can_not_ssh_to_private_ip(self):
 
146
        for x in xrange(1):
 
147
            try:
 
148
                conn = self.connect_ssh(self.data['private_ip'], TEST_KEY)
 
149
                conn.close()
 
150
            except Exception:
 
151
                time.sleep(1)
 
152
            else:
 
153
                self.fail('can ssh for ipv4 address from public network')
 
154
 
 
155
        if FLAGS.use_ipv6:
 
156
            for x in xrange(1):
 
157
                try:
 
158
                    conn = self.connect_ssh(
 
159
                                        self.data['ip_v6'], TEST_KEY)
 
160
                    conn.close()
 
161
                except Exception:
 
162
                    time.sleep(1)
 
163
                else:
 
164
                    self.fail('can ssh for ipv6 address from public network')
 
165
 
 
166
    def test_999_tearDown(self):
 
167
        self.delete_key_pair(self.conn, TEST_KEY)
 
168
        security_group_name = self.data['security_group_name']
 
169
        group = self.data['group']
 
170
        if group:
 
171
            group.revoke('tcp', 22, 22, '0.0.0.0/0')
 
172
            if FLAGS.use_ipv6:
 
173
                group.revoke('tcp', 22, 22, '::/0')
 
174
        self.conn.delete_security_group(security_group_name)
 
175
        if 'instance_id' in self.data:
 
176
            self.conn.terminate_instances([self.data['instance_id']])
 
177
 
 
178
if __name__ == "__main__":
 
179
    suites = {'instance': unittest.makeSuite(InstanceTestsFromPublic)}
 
180
    sys.exit(base.run_tests(suites))