1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2010 United States Government as represented by the
4
# Administrator of the National Aeronautics and Space Administration.
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
11
# http://www.apache.org/licenses/LICENSE-2.0
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
27
from smoketests import flags
28
from smoketests import base
29
from smoketests import user_smoketests
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
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
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)
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,
55
group.connection = self.conn
56
group.authorize('tcp', 22, 22, '0.0.0.0/0')
58
group.authorize('tcp', 22, 22, '::/0')
60
reservation = self.conn.run_instances(FLAGS.test_image,
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
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
74
if instance.state == u'running':
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
83
ipv6 = reservations[0].instances[0].dns_name_v6
84
self.failIf(ipv6 is None)
85
self.data['ip_v6'] = ipv6
87
def test_004_can_ssh_to_ipv6(self):
91
conn = self.connect_ssh(
92
self.data['ip_v6'], TEST_KEY)
94
except Exception as ex:
100
self.fail('could not ssh to instance')
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,
107
instance_type='m1.tiny')
108
self.assertEqual(len(reservation.instances), 1)
109
self.data['instance_id'] = reservation.instances[0].id
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
117
if instance.state == u'running':
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
126
ipv6 = reservations[0].instances[0].dns_name_v6
127
self.failIf(ipv6 is None)
128
self.data['ip_v6'] = ipv6
130
def test_014_can_not_ping_private_ip(self):
132
# ping waits for 1 second
133
status, output = commands.getstatusoutput(
134
'ping -c1 %s' % self.data['private_ip'])
136
self.fail('can ping private ip from public network')
138
status, output = commands.getstatusoutput(
139
'ping6 -c1 %s' % self.data['ip_v6'])
141
self.fail('can ping ipv6 from public network')
145
def test_015_can_not_ssh_to_private_ip(self):
148
conn = self.connect_ssh(self.data['private_ip'], TEST_KEY)
153
self.fail('can ssh for ipv4 address from public network')
158
conn = self.connect_ssh(
159
self.data['ip_v6'], TEST_KEY)
164
self.fail('can ssh for ipv6 address from public network')
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']
171
group.revoke('tcp', 22, 22, '0.0.0.0/0')
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']])
178
if __name__ == "__main__":
179
suites = {'instance': unittest.makeSuite(InstanceTestsFromPublic)}
180
sys.exit(base.run_tests(suites))