# -*- coding: utf-8 -*- # import Python Libs from __future__ import absolute_import from copy import deepcopy # import Python Third Party Libs try: import boto import boto.ec2.elb HAS_BOTO = True except ImportError: HAS_BOTO = False try: from moto import mock_ec2, mock_elb HAS_MOTO = True except ImportError: HAS_MOTO = False def mock_ec2(self): ''' if the mock_ec2 function is not available due to import failure this replaces the decorated function with stub_function. Allows boto_vpc unit tests to use the @mock_ec2 decorator without a "NameError: name 'mock_ec2' is not defined" error. ''' def stub_function(self): pass return stub_function def mock_elb(self): ''' if the mock_ec2 function is not available due to import failure this replaces the decorated function with stub_function. Allows boto_vpc unit tests to use the @mock_ec2 decorator without a "NameError: name 'mock_ec2' is not defined" error. ''' def stub_function(self): pass return stub_function # Import Salt Libs from salt.modules import boto_elb # Import Salt Testing Libs from salttesting import skipIf, TestCase from salttesting.mock import NO_MOCK, NO_MOCK_REASON from salttesting.helpers import ensure_in_syspath ensure_in_syspath('../../') region = 'us-east-1' access_key = 'GKTADJGHEIQSXMKKRBJ08H' secret_key = 'askdjghsdfjkghWupUjasdflkdfklgjsdfjajkghs' conn_parameters = {'region': region, 'key': access_key, 'keyid': secret_key, 'profile': {}} boto_conn_parameters = {'aws_access_key_id': access_key, 'aws_secret_access_key': secret_key} instance_parameters = {'instance_type': 't1.micro'} @skipIf(NO_MOCK, NO_MOCK_REASON) @skipIf(HAS_BOTO is False, 'The boto module must be installed.') @skipIf(HAS_MOTO is False, 'The moto module must be installed.') class BotoElbTestCase(TestCase): ''' TestCase for salt.modules.boto_elb module ''' @mock_ec2 @mock_elb def test_register_instances_valid_id_result_true(self): ''' tests that given a valid instance id and valid ELB that register_instances returns True. ''' conn_ec2 = boto.ec2.connect_to_region(region, **boto_conn_parameters) conn_elb = boto.ec2.elb.connect_to_region(region, **boto_conn_parameters) zones = [zone.name for zone in conn_ec2.get_all_zones()] elb_name = 'TestRegisterInstancesValidIdResult' conn_elb.create_load_balancer(elb_name, zones, [(80, 80, 'http')]) reservations = conn_ec2.run_instances('ami-08389d60') register_result = boto_elb.register_instances(elb_name, reservations.instances[0].id, **conn_parameters) self.assertEqual(True, register_result) @mock_ec2 @mock_elb def test_register_instances_valid_id_string(self): ''' tests that given a string containing a instance id and valid ELB that register_instances adds the given instance to an ELB ''' conn_ec2 = boto.ec2.connect_to_region(region, **boto_conn_parameters) conn_elb = boto.ec2.elb.connect_to_region(region, **boto_conn_parameters) zones = [zone.name for zone in conn_ec2.get_all_zones()] elb_name = 'TestRegisterInstancesValidIdResult' conn_elb.create_load_balancer(elb_name, zones, [(80, 80, 'http')]) reservations = conn_ec2.run_instances('ami-08389d60') boto_elb.register_instances(elb_name, reservations.instances[0].id, **conn_parameters) load_balancer_refreshed = conn_elb.get_all_load_balancers(elb_name)[0] registered_instance_ids = [instance.id for instance in load_balancer_refreshed.instances] print load_balancer_refreshed.instances self.assertEqual([reservations.instances[0].id], registered_instance_ids) @mock_ec2 @mock_elb def test_deregister_instances_valid_id_result_true(self): ''' tests that given an valid id the boto_elb deregister_instances method removes exactly one of a number of ELB registered instances ''' conn_ec2 = boto.ec2.connect_to_region(region, **boto_conn_parameters) conn_elb = boto.ec2.elb.connect_to_region(region, **boto_conn_parameters) zones = [zone.name for zone in conn_ec2.get_all_zones()] elb_name = 'TestDeregisterInstancesValidIdResult' load_balancer = conn_elb.create_load_balancer(elb_name, zones, [(80, 80, 'http')]) reservations = conn_ec2.run_instances('ami-08389d60') load_balancer.register_instances(reservations.instances[0].id) deregister_result = boto_elb.deregister_instances(elb_name, reservations.instances[0].id, **conn_parameters) self.assertEqual(True, deregister_result) @mock_ec2 @mock_elb def test_deregister_instances_valid_id_string(self): ''' tests that given an valid id the boto_elb deregister_instances method removes exactly one of a number of ELB registered instances ''' conn_ec2 = boto.ec2.connect_to_region(region, **boto_conn_parameters) conn_elb = boto.ec2.elb.connect_to_region(region, **boto_conn_parameters) zones = [zone.name for zone in conn_ec2.get_all_zones()] elb_name = 'TestDeregisterInstancesValidIdString' load_balancer = conn_elb.create_load_balancer(elb_name, zones, [(80, 80, 'http')]) reservations = conn_ec2.run_instances('ami-08389d60', min_count=2) all_instance_ids = [instance.id for instance in reservations.instances] load_balancer.register_instances(all_instance_ids) boto_elb.deregister_instances(elb_name, reservations.instances[0].id, **conn_parameters) load_balancer_refreshed = conn_elb.get_all_load_balancers(elb_name)[0] expected_instances = deepcopy(all_instance_ids) expected_instances.remove(reservations.instances[0].id) actual_instances = [instance.id for instance in load_balancer_refreshed.instances] self.assertEqual(actual_instances, expected_instances) @mock_ec2 @mock_elb def test_deregister_instances_valid_id_list(self): ''' tests that given an valid ids in the form of a list that the boto_elb deregister_instances all members of the given list ''' conn_ec2 = boto.ec2.connect_to_region(region, **boto_conn_parameters) conn_elb = boto.ec2.elb.connect_to_region(region, **boto_conn_parameters) zones = [zone.name for zone in conn_ec2.get_all_zones()] elb_name = 'TestDeregisterInstancesValidIdList' load_balancer = conn_elb.create_load_balancer(elb_name, zones, [(80, 80, 'http')]) reservations = conn_ec2.run_instances('ami-08389d60', min_count=3) all_instance_ids = [instance.id for instance in reservations.instances] load_balancer.register_instances(all_instance_ids) # reservations.instances[:-1] refers to all instances except list # instance deregister_instances = [instance.id for instance in reservations.instances[:-1]] expected_instances = [reservations.instances[-1].id] boto_elb.deregister_instances(elb_name, deregister_instances, **conn_parameters) load_balancer_refreshed = conn_elb.get_all_load_balancers(elb_name)[0] actual_instances = [instance.id for instance in load_balancer_refreshed.instances] self.assertEqual(actual_instances, expected_instances) if __name__ == '__main__': from integration import run_tests run_tests(BotoElbTestCase, needs_daemon=False)