<s> from __future__ import absolute_import <EOL> from distutils . version import LooseVersion <EOL> from salttesting . unit import skipIf , TestCase <EOL> from salttesting . mock import NO_MOCK , NO_MOCK_REASON , patch , MagicMock <EOL> from salttesting . helpers import ensure_in_syspath <EOL> ensure_in_syspath ( '<STR_LIT>' ) <EOL> from salt . exceptions import SaltInvocationError <EOL> import salt . utils . boto <EOL> try : <EOL> import boto <EOL> import boto . exception <EOL> from boto . exception import BotoServerError <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> required_boto_version = '<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> service = '<STR_LIT>' <EOL> resource_name = '<STR_LIT>' <EOL> resource_id = '<STR_LIT>' <EOL> error_body = '''<STR_LIT>''' <EOL> no_error_body = '''<STR_LIT>''' <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> def _has_required_moto ( ) : <EOL> '''<STR_LIT>''' <EOL> if not HAS_MOTO : <EOL> return False <EOL> else : <EOL> import pkg_resources <EOL> if LooseVersion ( pkg_resources . get_distribution ( '<STR_LIT>' ) . version ) < LooseVersion ( '<STR_LIT>' ) : <EOL> return False <EOL> return True <EOL> class BotoUtilsTestCaseBase ( TestCase ) : <EOL> def setUp ( self ) : <EOL> salt . utils . boto . __context__ = { } <EOL> salt . utils . boto . __opts__ = { } <EOL> salt . utils . boto . __pillar__ = { } <EOL> salt . utils . boto . __salt__ = { '<STR_LIT>' : MagicMock ( return_value = '<STR_LIT>' ) } <EOL> class BotoUtilsCacheIdTestCase ( BotoUtilsTestCaseBase ) : <EOL> def test_set_and_get_with_no_auth_params ( self ) : <EOL> salt . utils . boto . cache_id ( service , resource_name , resource_id = resource_id ) <EOL> self . assertEqual ( salt . utils . boto . cache_id ( service , resource_name ) , resource_id ) <EOL> def test_set_and_get_with_explicit_auth_params ( self ) : <EOL> salt . utils . boto . cache_id ( service , resource_name , resource_id = resource_id , ** conn_parameters ) <EOL> self . assertEqual ( salt . utils . boto . cache_id ( service , resource_name , ** conn_parameters ) , resource_id ) <EOL> def test_set_and_get_with_different_region_returns_none ( self ) : <EOL> salt . utils . boto . cache_id ( service , resource_name , resource_id = resource_id , region = '<STR_LIT>' ) <EOL> self . assertEqual ( salt . utils . boto . cache_id ( service , resource_name , region = '<STR_LIT>' ) , None ) <EOL> def test_set_and_get_after_invalidation_returns_none ( self ) : <EOL> salt . utils . boto . cache_id ( service , resource_name , resource_id = resource_id ) <EOL> salt . utils . boto . cache_id ( service , resource_name , resource_id = resource_id , invalidate = True ) <EOL> self . assertEqual ( salt . utils . boto . cache_id ( service , resource_name ) , None ) <EOL> def test_partial ( self ) : <EOL> cache_id = salt . utils . boto . cache_id_func ( service ) <EOL> cache_id ( resource_name , resource_id = resource_id ) <EOL> self . assertEqual ( cache_id ( resource_name ) , resource_id ) <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 BotoUtilsGetConnTestCase ( BotoUtilsTestCaseBase ) : <EOL> @ mock_ec2 <EOL> def test_conn_is_cached ( self ) : <EOL> conn = salt . utils . boto . get_connection ( service , ** conn_parameters ) <EOL> self . assertTrue ( conn in salt . utils . boto . __context__ . values ( ) ) <EOL> @ mock_ec2 <EOL> def test_conn_is_cache_with_profile ( self ) : <EOL> conn = salt . utils . boto . get_connection ( service , profile = conn_parameters ) <EOL> self . assertTrue ( conn in salt . utils . boto . __context__ . values ( ) ) <EOL> @ mock_ec2 <EOL> def test_get_conn_with_no_auth_params_raises_invocation_error ( self ) : <EOL> with patch ( '<STR_LIT>' . format ( service ) , <EOL> side_effect = boto . exception . NoAuthHandlerFound ( ) ) : <EOL> with self . assertRaises ( SaltInvocationError ) : <EOL> salt . utils . boto . get_connection ( service ) <EOL> @ mock_ec2 <EOL> def test_get_conn_error_raises_command_execution_error ( self ) : <EOL> with patch ( '<STR_LIT>' . format ( service ) , <EOL> side_effect = BotoServerError ( <NUM_LIT> , '<STR_LIT>' , body = error_body ) ) : <EOL> with self . assertRaises ( BotoServerError ) : <EOL> salt . utils . boto . get_connection ( service ) <EOL> @ mock_ec2 <EOL> def test_partial ( self ) : <EOL> get_conn = salt . utils . boto . get_connection_func ( service ) <EOL> conn = get_conn ( ** conn_parameters ) <EOL> self . assertTrue ( conn in salt . utils . boto . __context__ . values ( ) ) <EOL> @ skipIf ( HAS_BOTO is False , '<STR_LIT>' ) <EOL> @ skipIf ( _has_required_boto ( ) is False , '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( required_boto_version ) ) <EOL> class BotoUtilsGetErrorTestCase ( BotoUtilsTestCaseBase ) : <EOL> def test_error_message ( self ) : <EOL> e = BotoServerError ( '<STR_LIT>' , '<STR_LIT>' , body = error_body ) <EOL> r = salt . utils . boto . get_error ( e ) <EOL> expected = { '<STR_LIT>' : { '<STR_LIT:code>' : '<STR_LIT>' , <EOL> '<STR_LIT:message>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' } , <EOL> '<STR_LIT:message>' : '<STR_LIT>' } <EOL> self . assertEqual ( r , expected ) <EOL> def test_exception_message_with_no_body ( self ) : <EOL> e = BotoServerError ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> r = salt . utils . boto . get_error ( e ) <EOL> expected = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:status>' : '<STR_LIT>' } , <EOL> '<STR_LIT:message>' : '<STR_LIT>' } <EOL> self . assertEqual ( r , expected ) <EOL> def test_exception_message_with_no_error_in_body ( self ) : <EOL> e = BotoServerError ( '<STR_LIT>' , '<STR_LIT>' , body = no_error_body ) <EOL> r = salt . utils . boto . get_error ( e ) <EOL> expected = { '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:status>' : '<STR_LIT>' } , <EOL> '<STR_LIT:message>' : '<STR_LIT>' } <EOL> self . assertEqual ( r , expected ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> from integration import run_tests <EOL> run_tests ( BotoUtilsGetConnTestCase , needs_daemon = False ) </s>
