<s> from __future__ import absolute_import <EOL> import random <EOL> import string <EOL> from copy import deepcopy <EOL> from distutils . version import LooseVersion <EOL> from salttesting . unit import skipIf , TestCase <EOL> from salttesting . mock import NO_MOCK , NO_MOCK_REASON <EOL> from salttesting . helpers import ensure_in_syspath <EOL> ensure_in_syspath ( '<STR_LIT>' ) <EOL> import salt . config <EOL> import salt . loader <EOL> from salt . ext . six . moves import range <EOL> try : <EOL> import boto <EOL> HAS_BOTO = True <EOL> except ImportError : <EOL> HAS_BOTO = False <EOL> try : <EOL> from moto import mock_ec2 <EOL> HAS_MOTO = True <EOL> except ImportError : <EOL> HAS_MOTO = False <EOL> def mock_ec2 ( self ) : <EOL> '''<STR_LIT>''' <EOL> def stub_function ( self ) : <EOL> pass <EOL> return stub_function <EOL> from salt . utils . odict import OrderedDict <EOL> from salt . modules import boto_secgroup <EOL> required_boto_version = '<STR_LIT>' <EOL> vpc_id = '<STR_LIT>' <EOL> region = '<STR_LIT>' <EOL> access_key = '<STR_LIT>' <EOL> secret_key = '<STR_LIT>' <EOL> conn_parameters = { '<STR_LIT>' : region , '<STR_LIT:key>' : access_key , '<STR_LIT>' : secret_key , '<STR_LIT>' : { } } <EOL> boto_conn_parameters = { '<STR_LIT>' : access_key , '<STR_LIT>' : secret_key } <EOL> opts = salt . config . DEFAULT_MASTER_OPTS <EOL> utils = salt . loader . utils ( opts , whitelist = [ '<STR_LIT>' ] ) <EOL> funcs = salt . loader . minion_mods ( opts , utils = utils ) <EOL> boto_secgroup . __salt__ = funcs <EOL> boto_secgroup . __utils__ = utils <EOL> boto_secgroup . __virtual__ ( ) <EOL> def _random_group_id ( ) : <EOL> group_id = '<STR_LIT>' . format ( random . randrange ( <NUM_LIT:2> ** <NUM_LIT:32> ) ) <EOL> return group_id <EOL> def _random_group_name ( ) : <EOL> group_name = '<STR_LIT>' . format ( '<STR_LIT>' . join ( ( random . choice ( string . ascii_lowercase ) ) for char in range ( <NUM_LIT:12> ) ) ) <EOL> return group_name <EOL> def _has_required_boto ( ) : <EOL> '''<STR_LIT>''' <EOL> if not HAS_BOTO : <EOL> return False <EOL> elif LooseVersion ( boto . __version__ ) < LooseVersion ( required_boto_version ) : <EOL> return False <EOL> else : <EOL> return True <EOL> @ skipIf ( NO_MOCK , NO_MOCK_REASON ) <EOL> @ skipIf ( HAS_BOTO is False , '<STR_LIT>' ) <EOL> @ skipIf ( HAS_MOTO is False , '<STR_LIT>' ) <EOL> @ skipIf ( _has_required_boto ( ) is False , '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( required_boto_version ) ) <EOL> class BotoSecgroupTestCase ( TestCase ) : <EOL> '''<STR_LIT>''' <EOL> def test__split_rules ( self ) : <EOL> '''<STR_LIT>''' <EOL> rules = [ OrderedDict ( [ ( '<STR_LIT>' , u'<STR_LIT>' ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , [ OrderedDict ( [ ( '<STR_LIT>' , u'<STR_LIT>' ) ] ) ] ) ] ) , <EOL> OrderedDict ( [ ( '<STR_LIT>' , u'<STR_LIT>' ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , <NUM_LIT> ) , ( '<STR_LIT>' , [ OrderedDict ( [ ( '<STR_LIT>' , u'<STR_LIT>' ) ] ) ] ) ] ) ] <EOL> split_rules = [ { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : u'<STR_LIT>' , '<STR_LIT>' : u'<STR_LIT>' } , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : u'<STR_LIT>' , '<STR_LIT>' : u'<STR_LIT>' } ] <EOL> self . assertEqual ( boto_secgroup . _split_rules ( rules ) , split_rules ) <EOL> @ mock_ec2 <EOL> def test_create_ec2_classic ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> boto_secgroup . create ( group_name , group_description , ** conn_parameters ) <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group_filter = { '<STR_LIT>' : group_name } <EOL> secgroup_created_group = conn . get_all_security_groups ( filters = group_filter ) <EOL> expected_create_result = [ group_name , <EOL> group_description , <EOL> None ] <EOL> secgroup_create_result = [ secgroup_created_group [ <NUM_LIT:0> ] . name , <EOL> secgroup_created_group [ <NUM_LIT:0> ] . description , <EOL> secgroup_created_group [ <NUM_LIT:0> ] . vpc_id ] <EOL> self . assertEqual ( expected_create_result , secgroup_create_result ) <EOL> @ mock_ec2 <EOL> def test_create_ec2_vpc ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> boto_secgroup . create ( group_name , group_description , vpc_id = vpc_id , ** conn_parameters ) <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group_filter = { '<STR_LIT>' : group_name , '<STR_LIT>' : vpc_id } <EOL> secgroup_created_group = conn . get_all_security_groups ( filters = group_filter ) <EOL> expected_create_result = [ group_name , group_description , vpc_id ] <EOL> secgroup_create_result = [ secgroup_created_group [ <NUM_LIT:0> ] . name , secgroup_created_group [ <NUM_LIT:0> ] . description , secgroup_created_group [ <NUM_LIT:0> ] . vpc_id ] <EOL> self . assertEqual ( expected_create_result , secgroup_create_result ) <EOL> @ skipIf ( True , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock_ec2 <EOL> def test_get_group_id_ec2_classic ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group_classic = conn . create_security_group ( name = group_name , <EOL> description = group_description ) <EOL> group_vpc = conn . create_security_group ( name = group_name , <EOL> description = group_description , <EOL> vpc_id = vpc_id ) <EOL> retrieved_group_id = boto_secgroup . get_group_id ( group_name , <EOL> ** conn_parameters ) <EOL> self . assertEqual ( group_classic . id , retrieved_group_id ) <EOL> @ skipIf ( True , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock_ec2 <EOL> def test_get_group_id_ec2_vpc ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group_classic = conn . create_security_group ( name = group_name , <EOL> description = group_description ) <EOL> group_vpc = conn . create_security_group ( name = group_name , <EOL> description = group_description , <EOL> vpc_id = vpc_id ) <EOL> retrieved_group_id = boto_secgroup . get_group_id ( group_name , group_vpc , <EOL> ** conn_parameters ) <EOL> self . assertEqual ( group_vpc . id , retrieved_group_id ) <EOL> @ mock_ec2 <EOL> def test_get_config_single_rule_group_name ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> ip_protocol = '<STR_LIT>' <EOL> from_port = <NUM_LIT> <EOL> to_port = <NUM_LIT> <EOL> cidr_ip = '<STR_LIT>' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group = conn . create_security_group ( name = group_name , description = group_name ) <EOL> group . authorize ( ip_protocol = ip_protocol , from_port = from_port , to_port = to_port , cidr_ip = cidr_ip ) <EOL> expected_get_config_result = OrderedDict ( [ ( '<STR_LIT:name>' , group . name ) , ( '<STR_LIT>' , group . id ) , ( '<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( '<STR_LIT:description>' , group . description ) , ( '<STR_LIT>' , { } ) , <EOL> ( '<STR_LIT>' , [ { '<STR_LIT>' : to_port , '<STR_LIT>' : from_port , <EOL> '<STR_LIT>' : ip_protocol , '<STR_LIT>' : cidr_ip } ] ) , <EOL> ( '<STR_LIT>' , [ ] ) ] ) <EOL> secgroup_get_config_result = boto_secgroup . get_config ( group_id = group . id , ** conn_parameters ) <EOL> self . assertEqual ( expected_get_config_result , secgroup_get_config_result ) <EOL> @ skipIf ( True , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock_ec2 <EOL> def test_exists_true_name_classic ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group_classic = conn . create_security_group ( group_name , group_description ) <EOL> group_vpc = conn . create_security_group ( group_name , group_description , vpc_id = vpc_id ) <EOL> salt_exists_result = boto_secgroup . exists ( name = group_name , ** conn_parameters ) <EOL> self . assertTrue ( salt_exists_result ) <EOL> @ skipIf ( True , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock_ec2 <EOL> def test_exists_false_name_classic ( self ) : <EOL> pass <EOL> @ mock_ec2 <EOL> def test_exists_true_name_vpc ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> conn . create_security_group ( group_name , group_description , vpc_id = vpc_id ) <EOL> salt_exists_result = boto_secgroup . exists ( name = group_name , vpc_id = vpc_id , ** conn_parameters ) <EOL> self . assertTrue ( salt_exists_result ) <EOL> @ mock_ec2 <EOL> def test_exists_false_name_vpc ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> salt_exists_result = boto_secgroup . exists ( group_name , vpc_id = vpc_id , ** conn_parameters ) <EOL> self . assertFalse ( salt_exists_result ) <EOL> @ mock_ec2 <EOL> def test_exists_true_group_id ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group = conn . create_security_group ( group_name , group_description ) <EOL> salt_exists_result = boto_secgroup . exists ( group_id = group . id , ** conn_parameters ) <EOL> self . assertTrue ( salt_exists_result ) <EOL> @ mock_ec2 <EOL> def test_exists_false_group_id ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_id = _random_group_id ( ) <EOL> salt_exists_result = boto_secgroup . exists ( group_id = group_id , ** conn_parameters ) <EOL> self . assertFalse ( salt_exists_result ) <EOL> @ skipIf ( True , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock_ec2 <EOL> def test_delete_group_ec2_classic ( self ) : <EOL> '''<STR_LIT>''' <EOL> group_name = _random_group_name ( ) <EOL> group_description = '<STR_LIT>' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> group_classic = conn . create_security_group ( name = group_name , description = group_description ) <EOL> group_vpc = conn . create_security_group ( name = group_name , description = group_description , vpc_id = vpc_id ) <EOL> all_groups = [ group . id for group in conn . get_all_security_groups ( ) ] <EOL> deleted = boto_secgroup . delete ( name = group_name , ** conn_parameters ) <EOL> expected_groups = deepcopy ( all_groups ) <EOL> expected_groups . remove ( group_classic . id ) <EOL> actual_groups = [ group . id for group in conn . get_all_security_groups ( ) ] <EOL> self . assertEqual ( expected_groups , actual_groups ) <EOL> @ skipIf ( True , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ mock_ec2 <EOL> def test_delete_group_name_ec2_vpc ( self ) : <EOL> pass <EOL> @ mock_ec2 <EOL> def test__get_conn_true ( self ) : <EOL> '''<STR_LIT>''' <EOL> conn = boto . ec2 . connect_to_region ( region , ** boto_conn_parameters ) <EOL> salt_conn = boto_secgroup . _get_conn ( ** conn_parameters ) <EOL> self . assertEqual ( conn . __class__ , salt_conn . __class__ ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from integration import run_tests <EOL> run_tests ( BotoSecgroupTestCase , needs_daemon = False ) </s>
