How to use the taskcat._client_factory.Boto3Cache function in taskcat

To help you get started, we’ve selected a few taskcat examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github aws-quickstart / taskcat / tests / test_template_params.py View on Github external
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
github aws-quickstart / taskcat / tests / test_client_factory.py View on Github external
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)
github aws-quickstart / taskcat / tests / test_template_params.py View on Github external
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
github aws-quickstart / taskcat / tests / test_client_factory.py View on Github external
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)
github aws-quickstart / taskcat / tests / test_client_factory.py View on Github external
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")
github aws-quickstart / taskcat / tests / test_config.py View on Github external
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()
github aws-quickstart / taskcat / taskcat / _config.py View on Github external
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,
github aws-quickstart / taskcat / taskcat / _config.py View on Github external
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
github aws-quickstart / taskcat / taskcat / _cli_modules / deploy.py View on Github external
):
        """
        :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 = (