Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _mock_client method in localstack

Best Python code snippet using localstack_python

test_aws_utils.py

Source:test_aws_utils.py

1"""2Copyright (c) Contributors to the Open 3D Engine Project.3For complete copyright and license terms please see the LICENSE at the root of this distribution.4SPDX-License-Identifier: Apache-2.0 OR MIT5"""6from typing import List7from unittest import TestCase8from unittest.mock import (ANY, call, MagicMock, patch)9from model import constants10from model.basic_resource_attributes import (BasicResourceAttributes, BasicResourceAttributesBuilder)11from utils import aws_utils12import pytest13from botocore.exceptions import (CredentialRetrievalError, PartialCredentialsError)14class TestAWSUtils(TestCase):15    """16    aws utils unit test cases17    TODO: add test cases once error handling is ready18    """19    _expected_account_id: str = "1234567890"20    _expected_region: str = "aws-global"21    _expected_bucket: str = "TestBucket"22    _expected_function: str = "TestFunction"23    _expected_table: str = "TestTable"24    _expected_stack: str = "TestStack"25    _expected_bucket_resource: BasicResourceAttributes = BasicResourceAttributesBuilder() \26        .build_type(constants.AWS_RESOURCE_TYPES[constants.AWS_RESOURCE_S3_BUCKET_INDEX]) \27        .build_name_id(_expected_bucket) \28        .build()29    def setUp(self) -> None:30        session_patcher: patch = patch("boto3.session.Session")31        self.addCleanup(session_patcher.stop)32        self._mock_session: MagicMock = session_patcher.start()33        self._mock_client: MagicMock = self._mock_session.return_value.client34        aws_utils.setup_default_session("default")35    def test_get_default_account_id_return_expected_account_id(self) -> None:36        mocked_sts_client: MagicMock = self._mock_client.return_value37        mocked_sts_client.get_caller_identity.return_value = {"Account": TestAWSUtils._expected_account_id}38        actual_account_id: str = aws_utils.get_default_account_id()39        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.STS_SERVICE_NAME)40        mocked_sts_client.get_caller_identity.assert_called_once()41        assert actual_account_id == TestAWSUtils._expected_account_id42    def test_get_default_account_id_raise_credential_retrieval_error(self) -> None:43        self._mock_session.return_value.client.side_effect=CredentialRetrievalError(provider="custom-process", error_msg="")44        with pytest.raises(RuntimeError) as error:45            aws_utils.get_default_account_id()46        assert str(error.value) == 'Error when retrieving credentials from custom-process: '47    def test_get_default_account_id_raise_partial_credentials_error(self) -> None:48        self._mock_session.return_value.client.side_effect=PartialCredentialsError(provider="custom-process", cred_var="access_key")49        with pytest.raises(RuntimeError) as error:50            aws_utils.get_default_account_id()51        assert str(error.value) == 'Partial credentials found in custom-process, missing: access_key'52    def test_get_default_region_return_expected_region_from_session(self) -> None:53        mocked_session = self._mock_session.return_value54        mocked_session.region_name = TestAWSUtils._expected_region55        actual_region: str = aws_utils.get_default_region()56        self._mock_session.assert_called_once()57        assert actual_region == TestAWSUtils._expected_region58    def test_get_default_region_return_expected_region_from_sts_client(self) -> None:59        mocked_session: MagicMock = self._mock_session.return_value60        mocked_session.region_name = ""61        mocked_sts_client: MagicMock = self._mock_client.return_value62        mocked_sts_client.meta.region_name = TestAWSUtils._expected_region63        actual_region: str = aws_utils.get_default_region()64        self._mock_session.assert_called_once()65        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.STS_SERVICE_NAME)66        assert actual_region == TestAWSUtils._expected_region67    def test_get_default_region_return_empty(self) -> None:68        mocked_session: MagicMock = self._mock_session.return_value69        mocked_session.region_name = ""70        mocked_sts_client: MagicMock = self._mock_client.return_value71        mocked_sts_client.meta.region_name = ""72        actual_region: str = aws_utils.get_default_region()73        self._mock_session.assert_called_once()74        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.STS_SERVICE_NAME)75        assert not actual_region76    def test_list_s3_buckets_return_empty_list(self) -> None:77        mocked_s3_client: MagicMock = self._mock_client.return_value78        mocked_s3_client.list_buckets.return_value = {"Buckets": {}}79        actual_buckets: List[str] = aws_utils.list_s3_buckets()80        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME)81        mocked_s3_client.list_buckets.assert_called_once()82        assert not actual_buckets83    def test_list_s3_buckets_return_empty_list_with_given_region(self) -> None:84        mocked_s3_client: MagicMock = self._mock_client.return_value85        mocked_s3_client.list_buckets.return_value = {"Buckets": []}86        actual_buckets: List[str] = aws_utils.list_s3_buckets(TestAWSUtils._expected_region)87        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME,88                                                  region_name=TestAWSUtils._expected_region)89        mocked_s3_client.list_buckets.assert_called_once()90        assert not actual_buckets91    def test_list_s3_buckets_return_empty_list_with_no_matching_region(self) -> None:92        expected_region: str = "us-east-1"93        mocked_s3_client: MagicMock = self._mock_client.return_value94        expected_buckets: List[str] = [f"{TestAWSUtils._expected_bucket}1", f"{TestAWSUtils._expected_bucket}2"]95        mocked_s3_client.list_buckets.return_value = {"Buckets": [{"Name": expected_buckets[0]},96                                                                  {"Name": expected_buckets[1]}]}97        mocked_s3_client.get_bucket_location.side_effect = [{"LocationConstraint": "us-east-2"},98                                                            {"LocationConstraint": "us-west-1"}]99        actual_buckets: List[str] = aws_utils.list_s3_buckets(expected_region)100        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME,101                                                  region_name=expected_region)102        mocked_s3_client.list_buckets.assert_called_once()103        assert not actual_buckets104    def test_list_s3_buckets_return_expected_buckets_matching_region(self) -> None:105        expected_region: str = "us-west-2"106        mocked_s3_client: MagicMock = self._mock_client.return_value107        expected_buckets: List[str] = [f"{TestAWSUtils._expected_bucket}1", f"{TestAWSUtils._expected_bucket}2"]108        mocked_s3_client.list_buckets.return_value = {"Buckets": [{"Name": expected_buckets[0]},109                                                                  {"Name": expected_buckets[1]}]}110        mocked_s3_client.get_bucket_location.side_effect = [{"LocationConstraint": "us-west-2"},111                                                            {"LocationConstraint": "us-west-1"}]112        actual_buckets: List[str] = aws_utils.list_s3_buckets(expected_region)113        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME,114                                                  region_name=expected_region)115        mocked_s3_client.list_buckets.assert_called_once()116        assert len(actual_buckets) == 1117        assert actual_buckets[0] == expected_buckets[0]118    def test_list_s3_buckets_return_expected_iad_buckets(self) -> None:119        expected_region: str = "us-east-1"120        mocked_s3_client: MagicMock = self._mock_client.return_value121        expected_buckets: List[str] = [f"{TestAWSUtils._expected_bucket}1", f"{TestAWSUtils._expected_bucket}2"]122        mocked_s3_client.list_buckets.return_value = {"Buckets": [{"Name": expected_buckets[0]},123                                                                  {"Name": expected_buckets[1]}]}124        mocked_s3_client.get_bucket_location.side_effect = [{"LocationConstraint": None},125                                                            {"LocationConstraint": "us-west-1"}]126        actual_buckets: List[str] = aws_utils.list_s3_buckets(expected_region)127        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.S3_SERVICE_NAME,128                                                  region_name=expected_region)129        mocked_s3_client.list_buckets.assert_called_once()130        assert len(actual_buckets) == 1131        assert actual_buckets[0] == expected_buckets[0]132    def test_list_lambda_functions_return_empty_list(self) -> None:133        mocked_lambda_client: MagicMock = self._mock_client.return_value134        mocked_paginator: MagicMock = MagicMock()135        mocked_lambda_client.get_paginator.return_value = mocked_paginator136        mocked_iterator: MagicMock = MagicMock()137        mocked_paginator.paginate.return_value = mocked_iterator138        mocked_iterator.__iter__.return_value = [{"Functions": []}]139        actual_functions: List[str] = aws_utils.list_lambda_functions()140        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.LAMBDA_SERVICE_NAME)141        mocked_lambda_client.get_paginator.assert_called_once_with(142            aws_utils.AWSConstants.LAMBDA_LIST_FUNCTIONS_API_NAME)143        mocked_paginator.paginate.assert_called_once_with(144            PaginationConfig={"PageSize": aws_utils._PAGINATION_PAGE_SIZE})145        assert not actual_functions146    def test_list_lambda_functions_return_expected_functions(self) -> None:147        mocked_lambda_client: MagicMock = self._mock_client.return_value148        mocked_paginator: MagicMock = MagicMock()149        mocked_lambda_client.get_paginator.return_value = mocked_paginator150        mocked_iterator: MagicMock = MagicMock()151        mocked_paginator.paginate.return_value = mocked_iterator152        expected_functions: List[str] = [f"{TestAWSUtils._expected_function}1", f"{TestAWSUtils._expected_function}2"]153        mocked_iterator.__iter__.return_value = [{"Functions": [{"FunctionName": expected_functions[0]},154                                                                {"FunctionName": expected_functions[1]}]}]155        actual_functions: List[str] = aws_utils.list_lambda_functions()156        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.LAMBDA_SERVICE_NAME)157        mocked_lambda_client.get_paginator.assert_called_once_with(158            aws_utils.AWSConstants.LAMBDA_LIST_FUNCTIONS_API_NAME)159        mocked_paginator.paginate.assert_called_once_with(160            PaginationConfig={"PageSize": aws_utils._PAGINATION_PAGE_SIZE})161        assert actual_functions == expected_functions162    def test_list_dynamodb_tables_return_empty_list(self) -> None:163        mocked_dynamodb_client: MagicMock = self._mock_client.return_value164        mocked_paginator: MagicMock = MagicMock()165        mocked_dynamodb_client.get_paginator.return_value = mocked_paginator166        mocked_iterator: MagicMock = MagicMock()167        mocked_paginator.paginate.return_value = mocked_iterator168        mocked_iterator.__iter__.return_value = [{"TableNames": []}]169        actual_functions: List[str] = aws_utils.list_dynamodb_tables()170        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.DYNAMODB_SERVICE_NAME)171        mocked_dynamodb_client.get_paginator.assert_called_once_with(172            aws_utils.AWSConstants.DYNAMODB_LIST_TABLES_API_NAME)173        mocked_paginator.paginate.assert_called_once_with(174            PaginationConfig={"PageSize": aws_utils._PAGINATION_PAGE_SIZE})175        assert not actual_functions176    def test_list_dynamodb_tables_return_expected_tables(self) -> None:177        mocked_dynamodb_client: MagicMock = self._mock_client.return_value178        mocked_paginator: MagicMock = MagicMock()179        mocked_dynamodb_client.get_paginator.return_value = mocked_paginator180        mocked_iterator: MagicMock = MagicMock()181        mocked_paginator.paginate.return_value = mocked_iterator182        expected_tables: List[str] = [f"{TestAWSUtils._expected_table}1", f"{TestAWSUtils._expected_table}2"]183        mocked_iterator.__iter__.return_value = [{"TableNames": expected_tables}]184        actual_tables: List[str] = aws_utils.list_dynamodb_tables()185        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.DYNAMODB_SERVICE_NAME)186        mocked_dynamodb_client.get_paginator.assert_called_once_with(187            aws_utils.AWSConstants.DYNAMODB_LIST_TABLES_API_NAME)188        mocked_paginator.paginate.assert_called_once_with(189            PaginationConfig={"PageSize": aws_utils._PAGINATION_PAGE_SIZE})190        assert actual_tables == expected_tables191    def test_list_cloudformation_stacks_return_empty_list(self) -> None:192        mocked_cloudformation_client: MagicMock = self._mock_client.return_value193        mocked_paginator: MagicMock = MagicMock()194        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator195        mocked_iterator: MagicMock = MagicMock()196        mocked_paginator.paginate.return_value = mocked_iterator197        mocked_iterator.__iter__.return_value = [{"StackSummaries": []}]198        actual_stacks: List[str] = aws_utils.list_cloudformation_stacks()199        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME)200        mocked_cloudformation_client.get_paginator.assert_called_once_with(201            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACKS_API_NAME)202        mocked_paginator.paginate.assert_called_once_with(203            StackStatusFilter=aws_utils.AWSConstants.CLOUDFORMATION_STACKS_STATUS_FILTERS)204        assert not actual_stacks205    def test_list_cloudformation_stacks_return_empty_list_with_given_region(self) -> None:206        mocked_cloudformation_client: MagicMock = self._mock_client.return_value207        mocked_paginator: MagicMock = MagicMock()208        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator209        mocked_iterator: MagicMock = MagicMock()210        mocked_paginator.paginate.return_value = mocked_iterator211        mocked_iterator.__iter__.return_value = [{"StackSummaries": []}]212        actual_stacks: List[str] = aws_utils.list_cloudformation_stacks(TestAWSUtils._expected_region)213        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME,214                                                  region_name=TestAWSUtils._expected_region)215        mocked_cloudformation_client.get_paginator.assert_called_once_with(216            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACKS_API_NAME)217        mocked_paginator.paginate.assert_called_once_with(218            StackStatusFilter=aws_utils.AWSConstants.CLOUDFORMATION_STACKS_STATUS_FILTERS)219        assert not actual_stacks220    def test_list_cloudformation_stacks_return_expected_stacks(self) -> None:221        mocked_cloudformation_client: MagicMock = self._mock_client.return_value222        mocked_paginator: MagicMock = MagicMock()223        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator224        mocked_iterator: MagicMock = MagicMock()225        mocked_paginator.paginate.return_value = mocked_iterator226        expected_stacks: List[str] = [f"{TestAWSUtils._expected_stack}1", f"{TestAWSUtils._expected_stack}2"]227        mocked_iterator.__iter__.return_value = [{"StackSummaries": [{"StackName": expected_stacks[0]},228                                                                     {"StackName": expected_stacks[1]}]}]229        actual_stacks: List[str] = aws_utils.list_cloudformation_stacks()230        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME)231        mocked_cloudformation_client.get_paginator.assert_called_once_with(232            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACKS_API_NAME)233        mocked_paginator.paginate.assert_called_once_with(234            StackStatusFilter=aws_utils.AWSConstants.CLOUDFORMATION_STACKS_STATUS_FILTERS)235        assert actual_stacks == expected_stacks236    def test_list_cloudformation_stack_resources_return_empty_list(self) -> None:237        mocked_cloudformation_client: MagicMock = self._mock_client.return_value238        mocked_paginator: MagicMock = MagicMock()239        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator240        mocked_iterator: MagicMock = MagicMock()241        mocked_iterator.resume_token = None242        mocked_paginator.paginate.return_value = mocked_iterator243        mocked_iterator.__iter__.return_value = [{"StackResourceSummaries": []}]244        actual_stack_resources: List[BasicResourceAttributes] = \245            aws_utils.list_cloudformation_stack_resources(TestAWSUtils._expected_stack)246        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME)247        mocked_cloudformation_client.get_paginator.assert_called_once_with(248            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACK_RESOURCES_API_NAME)249        mocked_paginator.paginate.assert_called_once_with(StackName=TestAWSUtils._expected_stack, PaginationConfig=ANY)250        assert not actual_stack_resources251    def test_list_cloudformation_stack_resources_return_empty_list_with_given_region(self) -> None:252        mocked_cloudformation_client: MagicMock = self._mock_client.return_value253        mocked_paginator: MagicMock = MagicMock()254        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator255        mocked_iterator: MagicMock = MagicMock()256        mocked_iterator.resume_token = None257        mocked_paginator.paginate.return_value = mocked_iterator258        mocked_iterator.__iter__.return_value = [{"StackResourceSummaries": []}]259        actual_stack_resources: List[BasicResourceAttributes] = \260            aws_utils.list_cloudformation_stack_resources(TestAWSUtils._expected_stack, TestAWSUtils._expected_region)261        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME,262                                                  region_name=TestAWSUtils._expected_region)263        mocked_cloudformation_client.get_paginator.assert_called_once_with(264            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACK_RESOURCES_API_NAME)265        mocked_paginator.paginate.assert_called_once_with(StackName=TestAWSUtils._expected_stack, PaginationConfig=ANY)266        assert not actual_stack_resources267    def test_list_cloudformation_stack_resources_return_empty_list_when_resource_has_invalid_attributes(self) -> None:268        mocked_cloudformation_client: MagicMock = self._mock_client.return_value269        mocked_paginator: MagicMock = MagicMock()270        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator271        mocked_iterator: MagicMock = MagicMock()272        mocked_iterator.resume_token = None273        mocked_paginator.paginate.return_value = mocked_iterator274        mocked_iterator.__iter__.return_value = [{"StackResourceSummaries": [275            {"DummyAttribute": "DummyValue"}]}]276        actual_stack_resources: List[BasicResourceAttributes] = \277            aws_utils.list_cloudformation_stack_resources(TestAWSUtils._expected_stack, TestAWSUtils._expected_region)278        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME,279                                                  region_name=TestAWSUtils._expected_region)280        mocked_cloudformation_client.get_paginator.assert_called_once_with(281            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACK_RESOURCES_API_NAME)282        mocked_paginator.paginate.assert_called_once_with(StackName=TestAWSUtils._expected_stack, PaginationConfig=ANY)283        assert not actual_stack_resources284    def test_list_cloudformation_stack_resources_return_expected_stack_resources(self) -> None:285        mocked_cloudformation_client: MagicMock = self._mock_client.return_value286        mocked_paginator: MagicMock = MagicMock()287        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator288        mocked_iterator: MagicMock = MagicMock()289        mocked_iterator.resume_token = None290        mocked_iterator.__iter__.return_value = [{"StackResourceSummaries": [291            {"ResourceType": TestAWSUtils._expected_bucket_resource.type,292             "PhysicalResourceId": TestAWSUtils._expected_bucket_resource.name_id}]}]293        mocked_paginator.paginate.return_value = mocked_iterator294        actual_stack_resources: List[BasicResourceAttributes] = aws_utils.list_cloudformation_stack_resources(295            TestAWSUtils._expected_stack)296        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME)297        mocked_cloudformation_client.get_paginator.assert_called_once_with(298            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACK_RESOURCES_API_NAME)299        mocked_paginator.paginate.assert_called_once_with(StackName=TestAWSUtils._expected_stack, PaginationConfig=ANY)300        assert actual_stack_resources == [TestAWSUtils._expected_bucket_resource]301    def test_list_cloudformation_stack_resources_return_expected_stack_resources_with_separate_iterator(self) -> None:302        mocked_cloudformation_client: MagicMock = self._mock_client.return_value303        mocked_paginator: MagicMock = MagicMock()304        mocked_cloudformation_client.get_paginator.return_value = mocked_paginator305        mocked_iterator1: MagicMock = MagicMock()306        expected_starting_token: str = "starting_token"307        mocked_iterator1.resume_token = expected_starting_token308        mocked_iterator1.__iter__.return_value = [{"StackResourceSummaries": [309            {"ResourceType": TestAWSUtils._expected_bucket_resource.type,310             "PhysicalResourceId": TestAWSUtils._expected_bucket_resource.name_id}]}]311        mocked_iterator2: MagicMock = MagicMock()312        mocked_iterator2.resume_token = None313        mocked_iterator2.__iter__.return_value = [{"StackResourceSummaries": [314            {"ResourceType": TestAWSUtils._expected_bucket_resource.type,315             "PhysicalResourceId": TestAWSUtils._expected_bucket_resource.name_id}]}]316        mocked_paginator.paginate.side_effect = [mocked_iterator1, mocked_iterator2]317        actual_stack_resources: List[BasicResourceAttributes] = aws_utils.list_cloudformation_stack_resources(318            TestAWSUtils._expected_stack)319        self._mock_client.assert_called_once_with(aws_utils.AWSConstants.CLOUDFORMATION_SERVICE_NAME)320        mocked_cloudformation_client.get_paginator.assert_called_once_with(321            aws_utils.AWSConstants.CLOUDFORMATION_LIST_STACK_RESOURCES_API_NAME)322        mocked_calls: List[call] = [323            call(StackName=TestAWSUtils._expected_stack, PaginationConfig={"MaxItems": ANY, "StartingToken": None}),324            call(StackName=TestAWSUtils._expected_stack,325                 PaginationConfig={"MaxItems": ANY, "StartingToken": expected_starting_token})]326        mocked_paginator.paginate.assert_has_calls(mocked_calls)327        assert actual_stack_resources == [TestAWSUtils._expected_bucket_resource,...

test_update_kubeconfig.py

Source:test_update_kubeconfig.py

1# Copyright 2018 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7# http://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13import glob14import os15import mock16import tempfile17import shutil18import sys19import botocore20from botocore.compat import OrderedDict21from awscli.testutils import unittest22from awscli.customizations.utils import uni_print23import awscli.customizations.eks.kubeconfig as kubeconfig24from awscli.customizations.eks.update_kubeconfig import (KubeconfigSelector,25                                                         EKSClient,26                                                         API_VERSION)27from awscli.customizations.eks.exceptions import (EKSError,28                                                  EKSClusterError)29from awscli.customizations.eks.ordered_yaml import ordered_yaml_load30from tests.functional.eks.test_util import get_testdata31from tests.functional.eks.test_util import (describe_cluster_response,32                                            describe_cluster_no_status_response,33                                            describe_cluster_creating_response,34                                            describe_cluster_deleting_response)35def generate_env_variable(files):36    """37    Generate a string which is an environment variable38    containing the absolute paths for each file in files39    :param files: The names of the files to put in the environment variable40    :type files: list41    """42    output = ""43    for file in files:44        if len(output) == 0:45            output = file46        else:47            output += os.path.pathsep + file48    return output49EXAMPLE_ARN = "arn:aws:eks:region:111222333444:cluster/ExampleCluster"50class TestKubeconfigSelector(unittest.TestCase):51    def setUp(self):52        self._validator = kubeconfig.KubeconfigValidator()53        self._loader = kubeconfig.KubeconfigLoader(self._validator)54    def assert_chosen_path(self,55                           env_variable,56                           path_in,57                           cluster_name,58                           chosen_path):59        selector = KubeconfigSelector(env_variable, path_in, 60                                                    self._validator,61                                                    self._loader)62        self.assertEqual(selector.choose_kubeconfig(cluster_name).path,63                         chosen_path)  64    def test_parse_env_variable(self):65        paths = [66            "",67            "",68            get_testdata("valid_bad_cluster"),69            get_testdata("valid_bad_cluster2"),70            "",71            get_testdata("valid_existing"),72            ""73        ]74        env_variable = generate_env_variable(paths)75        selector = KubeconfigSelector(env_variable, None, self._validator,76                                                          self._loader)77        self.assertEqual(selector._paths, [path for path in paths 78                                                if len(path) > 0])79    def test_choose_env_only(self):80        paths = [81            get_testdata("valid_simple"),82            get_testdata("valid_existing")83        ] + glob.glob(get_testdata("invalid_*")) + [84            get_testdata("valid_bad_context"),85            get_testdata("valid_no_user")86        ]87        env_variable = generate_env_variable(paths)88        self.assert_chosen_path(env_variable, 89                                None, 90                                EXAMPLE_ARN, 91                                get_testdata("valid_simple"))92    def test_choose_existing(self):93        paths = [94            get_testdata("valid_simple"),95            get_testdata("valid_existing")96        ] + glob.glob(get_testdata("invalid_*")) + [97            get_testdata("valid_bad_context"),98            get_testdata("valid_no_user"),99            get_testdata("output_single"),100            get_testdata("output_single_with_role")101        ]102        env_variable = generate_env_variable(paths)103        self.assert_chosen_path(env_variable, 104                                None, 105                                EXAMPLE_ARN, 106                                get_testdata("output_single"))107    def test_arg_override(self):108        paths = [109            get_testdata("valid_simple"),110            get_testdata("valid_existing")111        ] + glob.glob(get_testdata("invalid_*")) + [112            get_testdata("valid_bad_context"),113            get_testdata("valid_no_user"),114            get_testdata("output_single"),115            get_testdata("output_single_with_role")116        ]117        env_variable = generate_env_variable(paths)118        self.assert_chosen_path(env_variable, 119                                get_testdata("output_combined"), 120                                EXAMPLE_ARN, 121                                get_testdata("output_combined"))122    def test_first_corrupted(self):123        paths = glob.glob(get_testdata("invalid_*")) + [124            get_testdata("valid_bad_context"),125            get_testdata("valid_no_user")126        ]127        env_variable = generate_env_variable(paths)128        selector = KubeconfigSelector(env_variable, None, self._validator,129                                                          self._loader)130        self.assertRaises(kubeconfig.KubeconfigCorruptedError, 131                          selector.choose_kubeconfig,132                          EXAMPLE_ARN)133    def test_arg_override_first_corrupted(self):134        paths = glob.glob(get_testdata("invalid_*")) + [135            get_testdata("valid_bad_context"),136            get_testdata("valid_no_user")137        ]138        env_variable = generate_env_variable(paths)139        self.assert_chosen_path(env_variable, 140                                get_testdata("output_combined"), 141                                EXAMPLE_ARN, 142                                get_testdata("output_combined"))143class TestEKSClient(unittest.TestCase):144    def setUp(self):145        self._correct_cluster_entry = OrderedDict([146            ("cluster", OrderedDict([147                ("certificate-authority-data", describe_cluster_response()\148                    ["cluster"]["certificateAuthority"]["data"]),149                ("server", describe_cluster_response()["cluster"]["endpoint"])150            ])),151            ("name", describe_cluster_response()["cluster"]["arn"])152        ])153        self._correct_user_entry = OrderedDict([154            ("name", describe_cluster_response()["cluster"]["arn"]),155            ("user", OrderedDict([156                ("exec", OrderedDict([157                    ("apiVersion", API_VERSION),158                    ("args",159                        [160                            "--region",161                            "region",162                            "eks",163                            "get-token",164                            "--cluster-name",165                            "ExampleCluster"166                        ]),167                    ("command", "aws")168                ]))169            ]))170        ])171        self._mock_client = mock.Mock()172        self._mock_client.describe_cluster.return_value =\173                                                    describe_cluster_response()174        self._session = mock.Mock(spec=botocore.session.Session)175        self._session.create_client.return_value = self._mock_client176        self._session.profile = None177        self._client = EKSClient(self._session, "ExampleCluster", None)178    def test_get_cluster_description(self):179        self.assertEqual(self._client._get_cluster_description(),180                         describe_cluster_response()["cluster"])181        self._mock_client.describe_cluster.assert_called_once_with(182            name="ExampleCluster"183        )184        self._session.create_client.assert_called_once_with("eks")185    def test_get_cluster_description_no_status(self):186        self._mock_client.describe_cluster.return_value = \187            describe_cluster_no_status_response()188        self.assertRaises(EKSClusterError,189                          self._client._get_cluster_description)190        self._mock_client.describe_cluster.assert_called_once_with(191            name="ExampleCluster"192        )193        self._session.create_client.assert_called_once_with("eks")194    def test_get_cluster_entry(self):195        self.assertEqual(self._client.get_cluster_entry(),196                         self._correct_cluster_entry)197        self._mock_client.describe_cluster.assert_called_once_with(198            name="ExampleCluster"199        )200        self._session.create_client.assert_called_once_with("eks")201    def test_get_user_entry(self):202        self.assertEqual(self._client.get_user_entry(),203                         self._correct_user_entry)204        self._mock_client.describe_cluster.assert_called_once_with(205            name="ExampleCluster"206        )207        self._session.create_client.assert_called_once_with("eks")208    def test_get_both(self):209        self.assertEqual(self._client.get_cluster_entry(),210                         self._correct_cluster_entry)211        self.assertEqual(self._client.get_user_entry(),212                         self._correct_user_entry)213        self._mock_client.describe_cluster.assert_called_once_with(214            name="ExampleCluster"215        )216        self._session.create_client.assert_called_once_with("eks")217    def test_cluster_creating(self):218        self._mock_client.describe_cluster.return_value =\219                                           describe_cluster_creating_response()220        self.assertRaises(EKSClusterError,221                          self._client._get_cluster_description)222        self._mock_client.describe_cluster.assert_called_once_with(223            name="ExampleCluster"224        )225        self._session.create_client.assert_called_once_with("eks")226    def test_cluster_deleting(self):227        self._mock_client.describe_cluster.return_value =\228                                           describe_cluster_deleting_response()229        self.assertRaises(EKSClusterError,230                          self._client._get_cluster_description)231        self._mock_client.describe_cluster.assert_called_once_with(232            name="ExampleCluster"233        )234        self._session.create_client.assert_called_once_with("eks")235    def test_profile(self):236        self._session.profile = "profile"237        self._correct_user_entry["user"]["exec"]["env"] = [238            OrderedDict([239                ("name", "AWS_PROFILE"),240                ("value", "profile")241            ])242        ]243        self.assertEqual(self._client.get_user_entry(),244                         self._correct_user_entry)245        self._mock_client.describe_cluster.assert_called_once_with(246            name="ExampleCluster"247        )...

conftest.py

Source:conftest.py

1import os2from typing import NamedTuple3from unittest.mock import Mock, sentinel4import pytest5import requests6import requests_mock7import vcr8from apiclient import APIClient9from apiclient.request_formatters import BaseRequestFormatter10from apiclient.response_handlers import BaseResponseHandler11BASE_DIR = os.path.abspath(os.path.realpath(os.path.dirname(__file__)))12VCR_CASSETTE_DIR = os.path.join(BASE_DIR, "vcr_cassettes")13api_client_vcr = vcr.VCR(14    serializer="yaml",15    cassette_library_dir=VCR_CASSETTE_DIR,16    record_mode="once",17    match_on=["uri", "method", "query"],18)19error_cassette_vcr = vcr.VCR(20    serializer="yaml", cassette_library_dir=VCR_CASSETTE_DIR, record_mode="once", match_on=["uri"]21)22@pytest.fixture23def cassette():24    with api_client_vcr.use_cassette("cassette.yaml") as cassette:25        yield cassette26@pytest.fixture27def error_cassette():28    with error_cassette_vcr.use_cassette("error_cassette.yaml") as cassette:29        yield cassette30@pytest.fixture31def mock_requests() -> requests_mock.Mocker:32    with requests_mock.mock() as _mocker:33        yield _mocker34class MockClient(NamedTuple):35    client: Mock36    request_formatter: Mock37    response_handler: Mock38@pytest.fixture39def mock_client():40    # Build our fully mocked client41    _mock_client: APIClient = Mock(spec=APIClient)42    mock_request_formatter: BaseRequestFormatter = Mock(spec=BaseRequestFormatter)43    mock_response_handler: BaseResponseHandler = Mock(spec=BaseResponseHandler)44    _mock_client.get_default_query_params.return_value = {}45    _mock_client.get_default_headers.return_value = {}46    _mock_client.get_default_username_password_authentication.return_value = None47    _mock_client.get_request_timeout.return_value = 30.048    _mock_client.get_session.return_value = requests.session()49    mock_request_formatter.format.return_value = {}50    _mock_client.get_request_formatter.return_value = mock_request_formatter51    mock_response_handler.get_request_data.return_value = sentinel.result52    _mock_client.get_response_handler.return_value = mock_response_handler53    return MockClient(54        client=_mock_client, request_formatter=mock_request_formatter, response_handler=mock_response_handler...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.