Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def record_flight_data(self, test_case, zdata=False, augment=False):
self.recording = True
test_dir = os.path.join(self.placebo_dir, test_case)
if not (zdata or augment):
if os.path.exists(test_dir):
shutil.rmtree(test_dir)
os.makedirs(test_dir)
session = boto3.Session()
default_region = session.region_name
if not zdata:
pill = placebo.attach(session, test_dir)
else:
pill = attach(session, self.archive_path, test_case, debug=True)
pill.record()
self.pill = pill
self.addCleanup(pill.stop)
self.addCleanup(self.cleanUp)
def factory(region=None, assume=None):
if region and region != default_region:
new_session = boto3.Session(region_name=region)
assert not zdata
new_pill = placebo.attach(new_session, test_dir, debug=True)
new_pill.record()
self.addCleanup(new_pill.stop)
return new_session
'region_name': os.environ.get('AWS_DEFAULT_REGION', 'us-east-1')
}
profile_name = os.environ.get('PLACEBO_PROFILE', None)
if profile_name:
session_kwargs['profile_name'] = profile_name
session = boto3.Session(**session_kwargs)
self = args[0]
prefix = self.__class__.__name__ + '.' + function.__name__
record_dir = os.path.join(PLACEBO_DIR, prefix)
if not os.path.exists(record_dir):
os.makedirs(record_dir)
pill = placebo.attach(session, data_path=record_dir)
if os.environ.get('PLACEBO_MODE') == 'record':
pill.record()
else:
pill.playback()
kwargs['session'] = session
return function(*args, **kwargs)
The `region` argument is to allow functional tests to override the
default region. It is unused when replaying stored data.
"""
if strtobool(os.environ.get('C7N_FUNCTIONAL', 'no')):
self.recording = True
return lambda region=region, assume=None: boto3.Session(region_name=region)
if not zdata:
test_dir = os.path.join(self.placebo_dir, test_case)
if not os.path.exists(test_dir):
raise RuntimeError("Invalid Test Dir for flight data %s" % test_dir)
session = boto3.Session()
if not zdata:
pill = placebo.attach(session, test_dir)
# pill = BluePill()
# pill.attach(session, test_dir)
else:
pill = attach(session, self.archive_path, test_case, False)
pill.playback()
self.addCleanup(pill.stop)
self.addCleanup(self.cleanUp)
return lambda region=None, assume=None: session
def setUp(self):
self.environ = {}
self.environ_patch = mock.patch('os.environ', self.environ)
self.environ_patch.start()
credential_path = os.path.join(os.path.dirname(__file__), 'cfg',
'aws_credentials')
self.environ['AWS_SHARED_CREDENTIALS_FILE'] = credential_path
self.data_path = os.path.join(os.path.dirname(__file__), 'responses')
self.data_path = os.path.join(self.data_path, 'tests')
self.session = boto3.Session(profile_name='foobar',
region_name='us-west-2')
self.pill = placebo.attach(self.session, self.data_path)
def setUp(self):
self.environ = {}
self.environ_patch = mock.patch('os.environ', self.environ)
self.environ_patch.start()
credential_path = os.path.join(os.path.dirname(__file__), 'cfg',
'aws_credentials')
self.environ['AWS_SHARED_CREDENTIALS_FILE'] = credential_path
self.data_path = os.path.join(os.path.dirname(__file__), 'responses')
self.data_path = os.path.join(self.data_path, 'saved')
self.session = boto3.Session(profile_name='foobar',
region_name='us-west-2')
self.pill = placebo.attach(self.session, self.data_path)
def test_assumed_session(self):
factory = self.replay_flight_data("test_credential_sts")
session = assumed_session(
role_arn='arn:aws:iam::644160558196:role/CustodianGuardDuty',
session_name="custodian-dev",
session=factory(),
)
# attach the placebo flight recorder to the new session.
pill = placebo.attach(
session, os.path.join(self.placebo_dir, 'test_credential_sts'))
if self.recording:
pill.record()
else:
pill.playback()
self.addCleanup(pill.stop)
try:
identity = session.client("sts").get_caller_identity()
except ClientError as e:
self.assertEqual(e.response["Error"]["Code"], "ValidationError")
self.assertEqual(
identity['Arn'],
'arn:aws:sts::644160558196:assumed-role/CustodianGuardDuty/custodian-dev')
profile_name = os.environ.get('PLACEBO_PROFILE', None)
if profile_name:
session_kwargs['profile_name'] = profile_name
session = boto3.Session(**session_kwargs)
self = args[0]
prefix = self.__class__.__name__ + '.' + function.__name__
base_dir = os.environ.get(
"PLACEBO_DIR", os.path.join(os.getcwd(), "placebo"))
record_dir = os.path.join(base_dir, prefix)
if not os.path.exists(record_dir):
os.makedirs(record_dir)
pill = placebo.attach(session, data_path=record_dir)
if os.environ.get('PLACEBO_MODE') == 'record':
pill.record()
else:
pill.playback()
kwargs['session'] = session
return function(*args, **kwargs)