diff --git a/test/test_auto_registration.py b/test/test_auto_registration.py index 4168fa9bc1..734090cbd2 100644 --- a/test/test_auto_registration.py +++ b/test/test_auto_registration.py @@ -47,50 +47,59 @@ AWS_TOKEN = "ABCDEFGHIJKLMNOPQRSTVWXYZabcdefghijklmnopqrstvwxyz0123==" -def send_only_imds_v2_is_supported(request, *args, **kwargs): +def send__aws_imdsv2_only(request, *args, **kwargs): """ - Mock result, when we try to get metadata using GET method against - AWS metadata provider. This mock is for the case, when only IMDSv2 - is supported by instance. + Mock result for metadata request on AWS where only IMDSv2 is supported. + This function should be used to replace function `requests.Session.send()`. + :param request: HTTP request - :return: Mock with result + :return: Mocked server result. """ - mock_result = Mock() + result = Mock() if request.method == "PUT": - if request.url == aws.AWSCloudProvider.CLOUD_PROVIDER_TOKEN_URL: - if "X-aws-ec2-metadata-token-ttl-seconds" in request.headers: - mock_result.status_code = 200 - mock_result.text = AWS_TOKEN - else: - mock_result.status_code = 400 - mock_result.text = "Error: TTL for token not specified" - else: - mock_result.status_code = 400 - mock_result.text = "Error: Invalid URL" - elif request.method == "GET": - if "X-aws-ec2-metadata-token" in request.headers.keys(): - if request.headers["X-aws-ec2-metadata-token"] == AWS_TOKEN: - if request.url == aws.AWSCloudProvider.CLOUD_PROVIDER_METADATA_URL: - mock_result.status_code = 200 - mock_result.text = AWS_METADATA - elif request.url == aws.AWSCloudProvider.CLOUD_PROVIDER_SIGNATURE_URL: - mock_result.status_code = 200 - mock_result.text = AWS_SIGNATURE - else: - mock_result.status_code = 400 - mock_result.text = "Error: Invalid URL" - else: - mock_result.status_code = 400 - mock_result.text = "Error: Invalid metadata token provided" - else: - mock_result.status_code = 400 - mock_result.text = "Error: IMDSv1 is not supported on this instance" - else: - mock_result.status_code = 400 - mock_result.text = "Error: not supported request method" - - return mock_result + if request.url != aws.AWSCloudProvider.CLOUD_PROVIDER_TOKEN_URL: + result.status_code = 400 + result.text = "Error: Invalid URL" + return result + + if "X-aws-ec2-metadata-token-ttl-seconds" not in request.headers: + result.status_code = 400 + result.text = "Error: TTL for token not specified" + return result + + result.status_code = 200 + result.text = AWS_TOKEN + return result + + if request.method == "GET": + if "X-aws-ec2-metadata-token" not in request.headers.keys(): + result.status_code = 400 + result.text = "Error: IMDSv1 is not supported on this instance" + return result + + if request.headers["X-aws-ec2-metadata-token"] != AWS_TOKEN: + result.status_code = 400 + result.text = "Error: Invalid metadata token provided" + return result + + if request.url == aws.AWSCloudProvider.CLOUD_PROVIDER_METADATA_URL: + result.status_code = 200 + result.text = AWS_METADATA + return result + + if request.url == aws.AWSCloudProvider.CLOUD_PROVIDER_SIGNATURE_URL: + result.status_code = 200 + result.text = AWS_SIGNATURE + return result + + result.status_code = 400 + result.text = "Error: Invalid URL" + return result + + result.status_code = 400 + result.text = "Error: not supported request method" + return result class TestAutomaticRegistration(unittest.TestCase): @@ -118,7 +127,7 @@ def test_collect_cloud_info_one_cloud_provider_detected(self): one detected cloud provider """ mock_session = Mock() - mock_session.send = send_only_imds_v2_is_supported() + mock_session.send = send__aws_imdsv2_only mock_session.prepare_request = Mock(side_effect=lambda request: request) mock_session.hooks = {"response": []} aws.AWSCloudProvider._instance._session = mock_session @@ -148,7 +157,7 @@ def test_collect_cloud_info_more_cloud_providers_detected(self): providers were detected """ mock_session = Mock() - mock_session.send = send_only_imds_v2_is_supported() + mock_session.send = send__aws_imdsv2_only mock_session.prepare_request = Mock(side_effect=lambda request: request) mock_session.hooks = {"response": []} aws.AWSCloudProvider._instance._session = mock_session