<s> import commands <EOL> import os <EOL> import random <EOL> import sys <EOL> import time <EOL> possible_topdir = os . path . normpath ( os . path . join ( os . path . abspath ( sys . argv [ <NUM_LIT:0> ] ) , <EOL> os . pardir , <EOL> os . pardir ) ) <EOL> if os . path . exists ( os . path . join ( possible_topdir , '<STR_LIT>' , '<STR_LIT>' ) ) : <EOL> sys . path . insert ( <NUM_LIT:0> , possible_topdir ) <EOL> from smoketests import flags <EOL> from smoketests import base <EOL> FLAGS = flags . FLAGS <EOL> TEST_PREFIX = '<STR_LIT>' % int ( random . random ( ) * <NUM_LIT> ) <EOL> TEST_BUCKET = '<STR_LIT>' % TEST_PREFIX <EOL> TEST_KEY = '<STR_LIT>' % TEST_PREFIX <EOL> TEST_KEY2 = '<STR_LIT>' % TEST_PREFIX <EOL> TEST_DATA = { } <EOL> class InstanceTestsFromPublic ( base . UserSmokeTestCase ) : <EOL> def test_001_can_create_keypair ( self ) : <EOL> key = self . create_key_pair ( self . conn , TEST_KEY ) <EOL> self . assertEqual ( key . name , TEST_KEY ) <EOL> def test_002_security_group ( self ) : <EOL> security_group_name = "<STR_LIT>" . join ( random . choice ( "<STR_LIT>" ) <EOL> for x in range ( random . randint ( <NUM_LIT:4> , <NUM_LIT:8> ) ) ) <EOL> group = self . conn . create_security_group ( security_group_name , <EOL> '<STR_LIT>' ) <EOL> group . connection = self . conn <EOL> group . authorize ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' ) <EOL> if FLAGS . use_ipv6 : <EOL> group . authorize ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' ) <EOL> reservation = self . conn . run_instances ( FLAGS . test_image , <EOL> key_name = TEST_KEY , <EOL> security_groups = [ security_group_name ] , <EOL> instance_type = '<STR_LIT>' ) <EOL> self . data [ '<STR_LIT>' ] = security_group_name <EOL> self . data [ '<STR_LIT>' ] = group <EOL> self . data [ '<STR_LIT>' ] = reservation . instances [ <NUM_LIT:0> ] . id <EOL> def test_003_instance_with_group_runs_within_60_seconds ( self ) : <EOL> reservations = self . conn . get_all_instances ( [ self . data [ '<STR_LIT>' ] ] ) <EOL> instance = reservations [ <NUM_LIT:0> ] . instances [ <NUM_LIT:0> ] <EOL> for x in xrange ( <NUM_LIT> ) : <EOL> instance . update ( ) <EOL> if instance . state == u'<STR_LIT>' : <EOL> break <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> ip = reservations [ <NUM_LIT:0> ] . instances [ <NUM_LIT:0> ] . private_dns_name <EOL> self . failIf ( ip == '<STR_LIT>' ) <EOL> self . data [ '<STR_LIT>' ] = ip <EOL> if FLAGS . use_ipv6 : <EOL> ipv6 = reservations [ <NUM_LIT:0> ] . instances [ <NUM_LIT:0> ] . dns_name_v6 <EOL> self . failIf ( ipv6 is None ) <EOL> self . data [ '<STR_LIT>' ] = ipv6 <EOL> def test_004_can_ssh_to_ipv6 ( self ) : <EOL> if FLAGS . use_ipv6 : <EOL> for x in xrange ( <NUM_LIT:20> ) : <EOL> try : <EOL> conn = self . connect_ssh ( <EOL> self . data [ '<STR_LIT>' ] , TEST_KEY ) <EOL> conn . close ( ) <EOL> except Exception as ex : <EOL> print ex <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> else : <EOL> break <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_012_can_create_instance_with_keypair ( self ) : <EOL> if '<STR_LIT>' in self . data : <EOL> self . conn . terminate_instances ( [ self . data [ '<STR_LIT>' ] ] ) <EOL> reservation = self . conn . run_instances ( FLAGS . test_image , <EOL> key_name = TEST_KEY , <EOL> instance_type = '<STR_LIT>' ) <EOL> self . assertEqual ( len ( reservation . instances ) , <NUM_LIT:1> ) <EOL> self . data [ '<STR_LIT>' ] = reservation . instances [ <NUM_LIT:0> ] . id <EOL> def test_013_instance_runs_within_60_seconds ( self ) : <EOL> reservations = self . conn . get_all_instances ( [ self . data [ '<STR_LIT>' ] ] ) <EOL> instance = reservations [ <NUM_LIT:0> ] . instances [ <NUM_LIT:0> ] <EOL> for x in xrange ( <NUM_LIT> ) : <EOL> instance . update ( ) <EOL> if instance . state == u'<STR_LIT>' : <EOL> break <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> ip = reservations [ <NUM_LIT:0> ] . instances [ <NUM_LIT:0> ] . private_dns_name <EOL> self . failIf ( ip == '<STR_LIT>' ) <EOL> self . data [ '<STR_LIT>' ] = ip <EOL> if FLAGS . use_ipv6 : <EOL> ipv6 = reservations [ <NUM_LIT:0> ] . instances [ <NUM_LIT:0> ] . dns_name_v6 <EOL> self . failIf ( ipv6 is None ) <EOL> self . data [ '<STR_LIT>' ] = ipv6 <EOL> def test_014_can_not_ping_private_ip ( self ) : <EOL> for x in xrange ( <NUM_LIT:4> ) : <EOL> status , output = commands . getstatusoutput ( <EOL> '<STR_LIT>' % self . data [ '<STR_LIT>' ] ) <EOL> if status == <NUM_LIT:0> : <EOL> self . fail ( '<STR_LIT>' ) <EOL> if FLAGS . use_ipv6 : <EOL> status , output = commands . getstatusoutput ( <EOL> '<STR_LIT>' % self . data [ '<STR_LIT>' ] ) <EOL> if status == <NUM_LIT:0> : <EOL> self . fail ( '<STR_LIT>' ) <EOL> else : <EOL> pass <EOL> def test_015_can_not_ssh_to_private_ip ( self ) : <EOL> for x in xrange ( <NUM_LIT:1> ) : <EOL> try : <EOL> conn = self . connect_ssh ( self . data [ '<STR_LIT>' ] , TEST_KEY ) <EOL> conn . close ( ) <EOL> except Exception : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> if FLAGS . use_ipv6 : <EOL> for x in xrange ( <NUM_LIT:1> ) : <EOL> try : <EOL> conn = self . connect_ssh ( <EOL> self . data [ '<STR_LIT>' ] , TEST_KEY ) <EOL> conn . close ( ) <EOL> except Exception : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> def test_999_tearDown ( self ) : <EOL> self . delete_key_pair ( self . conn , TEST_KEY ) <EOL> security_group_name = self . data [ '<STR_LIT>' ] <EOL> group = self . data [ '<STR_LIT>' ] <EOL> if group : <EOL> group . revoke ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' ) <EOL> if FLAGS . use_ipv6 : <EOL> group . revoke ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> , '<STR_LIT>' ) <EOL> self . conn . delete_security_group ( security_group_name ) <EOL> if '<STR_LIT>' in self . data : <EOL> self . conn . terminate_instances ( [ self . data [ '<STR_LIT>' ] ] ) </s>
