test_config.py

import os
import unittest
from pathlib import Path
from unittest import mock

from taskcat._client_factory import Boto3Cache
from taskcat._config import Config
from taskcat.exceptions import TaskCatException


class TestNewConfig(unittest.TestCase):
    def test_config(self):
        self.maxDiff = None
        base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
        base_path = Path(base_path + "data/config_inheritance").resolve()

        config = Config.create(
            args={"project": {"build_submodules": False}},
            global_config_path=base_path / ".taskcat_global.yml",
            project_config_path=base_path / "./.taskcat.yml",
            overrides_path=base_path / "./.taskcat_overrides.yml",
            env_vars={"TASKCAT_PROJECT_PACKAGE_LAMBDA": "False"},
        )

        expected = {
            "general": {
                "parameters": {
                    "GlobalVar": "set_in_global",
                    "OverridenVar": "set_in_global",
                },
                "s3_bucket": "set-in-global",
                "s3_regional_buckets": False,
            },
            "project": {
                "regions": ["us-east-1"],
                "package_lambda": False,
                "lambda_zip_path": "lambda_functions/packages",
                "lambda_source_path": "lambda_functions/source",
                "parameters": {
                    "GlobalVar": "set_in_global",
                    "OverridenVar": "set_in_global",
                    "ProjectVar": "set_in_project",
                },
                "build_submodules": False,
                "template": "template1.yaml",
                "s3_bucket": "set-in-global",
                "s3_enable_sig_v2": False,
                "s3_object_acl": "private",
                "shorten_stack_name": False,
                "s3_regional_buckets": False,
            },
            "tests": {
                "default": {
                    "parameters": {
                        "MyVar": "set_in_test",
                        "OverridenVar": "set_in_global",
                        "ProjectVar": "set_in_project",
                        "GlobalVar": "set_in_global",
                    },
                    "regions": ["us-west-2"],
                    "s3_bucket": "set-in-global",
                    "template": "template1.yaml",
                    "s3_regional_buckets": False,
                },
                "other": {
                    "template": "other_template.yaml",
                    "parameters": {
                        "ProjectVar": "set_in_project",
                        "OverridenVar": "set_in_global",
                        "GlobalVar": "set_in_global",
                    },
                    "regions": ["us-east-1"],
                    "s3_bucket": "set-in-global",
                    "s3_regional_buckets": False,
                },
            },
        }

        expected_source = {
            "general": {
                "s3_bucket": str(base_path / ".taskcat_global.yml"),
                "parameters": {
                    "GlobalVar": str(base_path / ".taskcat_global.yml"),
                    "OverridenVar": str(base_path / ".taskcat_global.yml"),
                },
                "s3_regional_buckets": str(base_path / ".taskcat_global.yml"),
            },
            "project": {
                "s3_bucket": str(base_path / ".taskcat_global.yml"),
                "s3_enable_sig_v2": "TASKCAT_DEFAULT",
                "s3_object_acl": "CliArgument",
                "s3_regional_buckets": str(base_path / ".taskcat_global.yml"),
                "shorten_stack_name": "TASKCAT_DEFAULT",
                "package_lambda": "EnvoronmentVariable",
                "lambda_zip_path": "TASKCAT_DEFAULT",
                "lambda_source_path": "TASKCAT_DEFAULT",
                "build_submodules": "CliArgument",
                "parameters": {
                    "GlobalVar": str(base_path / ".taskcat_global.yml"),
                    "OverridenVar": str(base_path / ".taskcat_global.yml"),
                    "ProjectVar": str(base_path / ".taskcat.yml"),
                },
                "regions": str(base_path / ".taskcat.yml"),
                "template": str(base_path / ".taskcat.yml"),
            },
            "tests": {
                "default": {
                    "s3_bucket": str(base_path / ".taskcat_global.yml"),
                    "s3_regional_buckets": str(base_path / ".taskcat_global.yml"),
                    "template": str(base_path / ".taskcat.yml"),
                    "parameters": {
                        "GlobalVar": str(base_path / ".taskcat_global.yml"),
                        "MyVar": str(base_path / ".taskcat.yml"),
                        "OverridenVar": str(base_path / ".taskcat_global.yml"),
                        "ProjectVar": str(base_path / ".taskcat.yml"),
                    },
                    "regions": str(base_path / ".taskcat.yml"),
                },
                "other": {
                    "s3_bucket": str(base_path / ".taskcat_global.yml"),
                    "s3_regional_buckets": str(base_path / ".taskcat_global.yml"),
                    "template": str(base_path / ".taskcat.yml"),
                    "parameters": {
                        "GlobalVar": str(base_path / ".taskcat_global.yml"),
                        "ProjectVar": str(base_path / ".taskcat.yml"),
                        "OverridenVar": str(base_path / ".taskcat_global.yml"),
                    },
                    "regions": str(base_path / ".taskcat.yml"),
                },
            },
        }

        self.assertEqual(expected, config.config.to_dict())
        self.assertEqual(expected_source, config.config._source)

    def test_legacy_config(self):

        base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
        base_path = Path(base_path + "data/legacy_test").resolve()

        new_config_location = base_path / ".taskcat.yml"
        new_overrides_location = base_path / ".taskcat_overrides.yml"

        if new_config_location.is_file():
            new_config_location.unlink()
        if new_overrides_location.is_file():
            new_overrides_location.unlink()

        Config.create(
            project_root=base_path,
            project_config_path=new_config_location,
            overrides_path=new_overrides_location,
        )

        self.assertTrue(new_config_location.is_file())
        self.assertTrue(new_overrides_location.is_file())

        # should not raise even if both legacy and current format files are present
        Config.create(
            project_root=base_path,
            project_config_path=new_config_location,
            overrides_path=new_overrides_location,
        )

    def test_standalone_template(self):

        base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
        base_path = Path(base_path + "data/legacy_test/templates/").resolve()

        Config.create(template_file=base_path / "test.template.yaml")

    def test_no_parameters(self):
        base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
        base_path = Path(base_path + "data/create_no_params/").resolve()

        Config.create(
            project_root=base_path, project_config_path=base_path / ".taskcat.yml"
        )

    @mock.patch("taskcat._config.Boto3Cache.account_id", return_value="123412341234")
    @mock.patch("taskcat._config.Boto3Cache.partition", return_value="aws")
    def test_get_regions(self, _, __):
        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={},
        )
        sessions = config.get_regions()
        for test_name, regions in sessions.items():
            with self.subTest(test=test_name):
                for region_name, region_obj in regions.items():
                    with self.subTest(region=region_name):
                        self.assertEqual(region_name, region_obj.name)
                        if test_name == "json-test" and region_name == "eu-central-1":
                            self.assertEqual("special-use-case", region_obj.profile)
                        elif test_name == "yaml-test" and region_name == "sa-east-1":
                            self.assertEqual("default", region_obj.profile)
                        elif region_name == "me-south-1":
                            self.assertEqual("mes1", region_obj.profile)
                        elif region_name == "ap-east-1":
                            self.assertEqual("hongkong", region_obj.profile)
                        elif test_name == "yaml-test":
                            self.assertEqual("foobar", region_obj.profile)
                        else:
                            self.assertEqual("default", region_obj.profile)

    @mock.patch("taskcat._config.Boto3Cache.account_id", return_value="123412341234")
    @mock.patch("taskcat._config.Boto3Cache.partition", return_value="aws")
    @mock.patch("taskcat._config.S3BucketObj.create", return_value=None)
    @mock.patch("taskcat._client_factory.boto3", autospec=True)
    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()

    @mock.patch("taskcat._config.Boto3Cache.account_id", return_value="123412341234")
    @mock.patch("taskcat._config.Boto3Cache.partition", return_value="aws")
    @mock.patch("taskcat._config.S3BucketObj.create", return_value=None)
    @mock.patch("taskcat._client_factory.boto3", autospec=True)
    def test_get_buckets_regional(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_regional_bucket.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)
        for test_name, regions in buckets.items():
            with self.subTest(test=test_name):
                for region_name, bucket_obj in regions.items():
                    self.assertEqual(bucket_obj.account_id, "123412341234")
                    self.assertEqual(bucket_obj.region, region_name)
                    self.assertTrue(bucket_obj.auto_generated)
                    self.assertTrue(bucket_obj.sigv4, True)
                    self.assertEqual(bucket_obj.partition, "aws")
                    self.assertEqual(
                        bucket_obj.name,
                        f"tcat-13725204b43e5bf5a37800c23614ee21-{region_name}",
                    )

    @mock.patch("taskcat._config.Boto3Cache.account_id", return_value="123412341234")
    @mock.patch("taskcat._config.Boto3Cache.partition", return_value="aws")
    @mock.patch("taskcat._config.S3BucketObj.create", return_value=None)
    @mock.patch(
        "taskcat._config.ParamGen._get_license_content_wrapper",
        return_value="a-license-key",
    )
    @mock.patch("taskcat._config.Boto3Cache", autospec=True)
    def test_get_rendered_params(self, _, __, ___, ____, m_boto):
        base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
        base_path = Path(base_path + "data/regional_client_and_bucket").resolve()
        m_boto.client.return_value = mock_client()
        config = Config.create(
            args={},
            project_root=base_path,
            global_config_path=base_path / ".taskcat_global.yml",
            project_config_path=base_path / "./.taskcat.yml",
            overrides_path=base_path / "./.taskcat_overrides.yml",
            env_vars={},
        )
        regions = config.get_regions(boto3_cache=m_boto)
        buckets = config.get_buckets(boto3_cache=m_boto)
        templates = config.get_templates()
        rendered_params = config.get_rendered_parameters(buckets, regions, templates)
        for test_name, regions in rendered_params.items():
            with self.subTest(test=test_name):
                for region_name, _params in regions.items():
                    with self.subTest(region=region_name):
                        buckets[test_name][region_name].delete()

    def test_get_templates(self):
        base_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
        base_path = Path(base_path + "data/regional_client_and_bucket").resolve()
        config = Config.create(
            args={},
            project_root=base_path,
            global_config_path=base_path / ".taskcat_global.yml",
            project_config_path=base_path / "./.taskcat.yml",
            overrides_path=base_path / "./.taskcat_overrides.yml",
            env_vars={},
        )
        templates = config.get_templates()
        for test_name, _template in templates.items():
            with self.subTest(test=test_name):
                pass

    @mock.patch("taskcat._config.LOG", autospec=True)
    def test__dict_from_template(self, mock_log):
        root_path = "./" if os.getcwd().endswith("/tests") else "./tests/"
        base_path = Path(root_path + "data/regional_client_and_bucket").resolve()
        template = base_path / "templates/debug-yaml.template"
        # valid config
        template_dict = Config._dict_from_template(template)
        self.assertEqual(True, isinstance(template_dict, dict))

        # invalid path
        with self.assertRaises(TaskCatException):
            Config._dict_from_template(base_path / "invalid-path")

        # cannot create template object
        with mock.patch("taskcat._config.Template") as mock_template:
            exc = ValueError("fail")
            mock_template.side_effect = exc
            with self.assertRaises(ValueError) as e:
                Config._dict_from_template(template)
            self.assertEqual(exc, e.exception)
            mock_log.warning.assert_called_once()

        base_path = Path(root_path + "data/standalone_template").resolve()

        # metadata in taskcat, but no taskcat key
        template = base_path / "test.template_no_tc_meta.yaml"
        template_dict = Config._dict_from_template(template)
        self.assertEqual(True, isinstance(template_dict, dict))
        self.assertEqual(
            True, template_dict["project"]["template"].endswith("no_tc_meta.yaml")
        )
        self.assertEqual({}, template_dict["tests"]["default"]["parameters"])

        # empty dict taskcat metadata
        template = base_path / "test.template_tc_empty_meta.yaml"
        template_dict = Config._dict_from_template(template)
        self.assertEqual(True, isinstance(template_dict, dict))
        self.assertEqual(
            True, template_dict["project"]["template"].endswith("tc_empty_meta.yaml")
        )
        self.assertEqual({}, template_dict["tests"]["default"]["parameters"])

        # populated taskcat metadata
        template = base_path / "test.template_tc_full_meta.yaml"
        template_dict = Config._dict_from_template(template)
        self.assertEqual(True, isinstance(template_dict, dict))
        self.assertEqual(
            True, template_dict["project"]["template"].endswith("tc_full_meta.yaml")
        )
        self.assertEqual(
            {"SomeParam": "SomeValue"}, template_dict["tests"]["sometest"]["parameters"]
        )


def mock_client(*args, **kwargs):
    m = mock.Mock()
    m.describe_availability_zones = mock_get_azs
    return m


def mock_get_azs(*args, **kwargs):
    return {
        "AvailabilityZones": [
            {"ZoneName": "mo-ck-1a", "ZoneId": "mockzoneid1a"},
            {"ZoneName": "mo-ck-1b", "ZoneId": "mockzoneid1b"},
            {"ZoneName": "mo-ck-1c", "ZoneId": "mockzoneid1c"},
        ]
    }