Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
