Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup_client_and_resource(self):
self._write_models()
self.loader = Loader(extra_search_paths=[self.root_dir])
self.botocore_session = botocore.session.get_session()
self.botocore_session.register_component('data_loader', self.loader)
self.session = Session(
botocore_session=self.botocore_session, region_name='us-east-1')
self.client = self.session.client('myservice', 'us-east-1')
self.resource = self.session.resource('myservice', 'us-east-1')
def test_can_create_deletion_deployer():
session = botocore.session.get_session()
deployer = create_deletion_deployer(TypedAWSClient(session), UI())
assert isinstance(deployer, Deployer)
def test_patch(region_name, service_name, make_test_session):
"""Test patching."""
localstack = make_test_session(region_name=region_name)
with localstack:
# Haven't patched yet.
# A regular botocore client should point to AWS right now.
original_bc_session = botocore.session.get_session()
original_bc_client = original_bc_session.create_client(
service_name, localstack.region_name
)
assert "127.0.0.1" not in original_bc_client._endpoint.host
with localstack.botocore.patch_botocore():
# Original client should now point to Localstack
assert "127.0.0.1" in original_bc_client._endpoint.host
# Original session should create Localstack clients
ls_client = original_bc_session.create_client(
service_name, localstack.region_name
)
assert "127.0.0.1" in ls_client._endpoint.host
# Original client back to AWS
def get_session():
mysess = session.get_session()
mysess.unregister('after-call.ec2.GetConsoleOutput',
handlers.decode_console_output)
mysess.register('after-call.ec2.GetConsoleOutput',
_decode_console_output_as_bytes)
return boto3.Session(botocore_session=mysess)
def main(self):
parser = self._create_parser()
self._parsed_args = parser.parse_args(self.args)
if self._parsed_args.profile:
self._session = Session(profile=self._parsed_args.profile)
else:
self._session = get_session()
if self._parsed_args.debug:
self._debug = True
if self._parsed_args.no_headless:
self._headless = False
if self._parsed_args.role:
self._role = self._parsed_args.role
if self._parsed_args.account:
self._account = self._parsed_args.account
return self.__getattribute__('_{}'.format(self._parsed_args.command))()
def _set_credentials(self):
'''
:param config_data: contents of the inventory config file
'''
self.boto_profile = self.get_option('aws_profile')
self.aws_access_key_id = self.get_option('aws_access_key')
self.aws_secret_access_key = self.get_option('aws_secret_key')
self.aws_security_token = self.get_option('aws_security_token')
self.iam_role_arn = self.get_option('iam_role_arn')
if not self.boto_profile and not (self.aws_access_key_id and self.aws_secret_access_key):
session = botocore.session.get_session()
try:
credentials = session.get_credentials().get_frozen_credentials()
except AttributeError:
pass
else:
self.aws_access_key_id = credentials.access_key
self.aws_secret_access_key = credentials.secret_key
self.aws_security_token = credentials.token
if not self.boto_profile and not (self.aws_access_key_id and self.aws_secret_access_key):
raise AnsibleError("Insufficient boto credentials found. Please provide them in your "
"inventory configuration file or set them as environment variables.")
def botocore_default_region(session: Optional[Session] = None) -> Optional[str]:
""" Returns default region name as configured on the system.
"""
if session is None:
session = botocore.session.get_session()
return session.get_config_variable('region')
def _probe_services(self):
try:
# create a config for all clients to share
client_config = botocore.config.Config(
connect_timeout=config.CLIENT_CONNECT_TIMEOUT,
read_timeout=config.CLIENT_READ_TIMEOUT
)
session = botocore.session.get_session()
for svc_name in self.svc_descriptors:
operations = self.svc_descriptors[svc_name]['ops']
regions = self.svc_descriptors[svc_name]['regions']
# call each API across each region
api_version = session.get_config_variable('api_versions').get(svc_name, None)
params = {'svc_name': svc_name,
'dry_run': self.script_args.dry_run,
'store': self.store}
for region in regions:
self.progress_bar.update_svc_text(svc_name, region)
try:
client = session.create_client(
svc_name,
region_name=region,
api_version=api_version,