Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def client_factory_instance():
with mock.patch.object(Boto3Cache, "__init__", return_value=None):
aws_clients = Boto3Cache(None)
aws_clients._credential_sets = {"default": [None, None, None, None]}
aws_clients.logger = logger
return aws_clients
def test_session_invalid_profile(self, mock_cache_lookup, mock_cache_set):
mock_cache_lookup.side_effect = ProfileNotFound(profile="non-existent-profile")
cache = Boto3Cache()
with self.assertRaises(ProfileNotFound):
cache.session(profile="non-existent-profile")
self.assertEqual(mock_cache_lookup.called, False)
cache._get_region = mock.Mock(return_value="us-east-1")
with self.assertRaises(ProfileNotFound):
cache.session(profile="non-existent-profile")
self.assertEqual(mock_cache_lookup.called, True)
def client_factory_instance():
with mock.patch.object(Boto3Cache, "__init__", return_value=None):
aws_clients = Boto3Cache(None)
aws_clients._credential_sets = {"default": [None, None, None, None]}
aws_clients.logger = logger
return aws_clients
def test_session_no_profile(self, mock_boto3, mock_cache_lookup, mock_cache_set):
mock_cache_lookup.side_effect = ProfileNotFound(profile="non-existent-profile")
Boto3Cache().session() # default value should be "default" profile
self.assertEqual(mock_boto3.called, True)
self.assertEqual(mock_cache_lookup.called, True)
self.assertEqual(mock_cache_set.called, True)
def test_stable_concurrency(self, mock_boto3):
# Sometimes boto fails with KeyErrors under high concurrency
for key_error in ["endpoint_resolver", "credential_provider"]:
mock_boto3.Session.side_effect = [KeyError(key_error), mock.DEFAULT]
c = Boto3Cache(_boto3=mock_boto3)
c.session("default")
def test_get_buckets(self, _, __, ___, m_boto):
base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
base_path = Path(base_path + "data/regional_client_and_bucket").resolve()
config = Config.create(
args={},
global_config_path=base_path / ".taskcat_global.yml",
project_config_path=base_path / "./.taskcat.yml",
overrides_path=base_path / "./.taskcat_overrides.yml",
env_vars={},
)
mock_boto_cache = Boto3Cache(_boto3=m_boto)
buckets = config.get_buckets(boto3_cache=mock_boto_cache)
bucket_acct = {}
for test_name, regions in buckets.items():
with self.subTest(test=test_name):
for region_name, region_obj in regions.items():
with self.subTest(region=region_name):
if not bucket_acct.get(region_obj.account_id):
bucket_acct[region_obj.account_id] = region_obj.name
self.assertEqual(
bucket_acct[region_obj.account_id], region_obj.name
)
region_obj.delete()
def get_regions(self, boto3_cache: Boto3Cache = None):
if boto3_cache is None:
boto3_cache = Boto3Cache()
region_objects: Dict[str, Dict[str, RegionObj]] = {}
for test_name, test in self.config.tests.items():
region_objects[test_name] = {}
for region in test.regions:
default_auth = (
test.auth.get("default", "default") if test.auth else "default"
)
profile = (
test.auth.get(region, default_auth) if test.auth else default_auth
)
region_objects[test_name][region] = RegionObj(
name=region,
account_id=boto3_cache.account_id(profile),
partition=boto3_cache.partition(profile),
profile=profile,
def get_regions(self, boto3_cache: Boto3Cache = None):
if boto3_cache is None:
boto3_cache = Boto3Cache()
region_objects: Dict[str, Dict[str, RegionObj]] = {}
for test_name, test in self.config.tests.items():
region_objects[test_name] = {}
for region in test.regions:
profile = test.auth.get(region, "default") if test.auth else "default"
region_objects[test_name][region] = RegionObj(
name=region,
account_id=boto3_cache.account_id(profile),
partition=boto3_cache.partition(profile),
profile=profile,
_boto3_cache=boto3_cache,
taskcat_id=self.uid,
)
return region_objects
):
"""
:param package: name of package to install can be a path to a local package,
a github org/repo, or an AWS Quick Start name
:param aws_profile: aws profile to use for installation
:param region: regions to install into, default will use aws cli configured
default
:param parameters: parameters to pass to the stack, in the format
Key=Value,AnotherKey=AnotherValue or providing a path to a json or yaml file
containing the parameters
:param name: stack name to use, if not specified one will be automatically
generated
:param wait: if enabled, taskcat will wait for stack to complete before exiting
"""
LOG.warning("deploy is in alpha feature, use with caution")
boto3_cache = Boto3Cache()
if not name:
name = generate_name()
if region == "default":
region = boto3_cache.get_default_region(profile_name=aws_profile)
path = Path(package).resolve()
if Path(package).resolve().is_dir():
package_type = "local"
elif "/" in package:
package_type = "github"
else: # assuming it's an AWS Quick Start
package_type = "github"
package = f"aws-quickstart/quickstart-{package}"
if package_type == "github":
if package.startswith("https://") or package.startswith("git@"):
url = package
org, repo = (