Skip to content

Commit

Permalink
Simplify autoregistration test setup
Browse files Browse the repository at this point in the history
The function responsible for returning 'server' responses was altered to
be less nested, which should improve its readability.

Cherry-picked from 4d8fcd4.
  • Loading branch information
m-horky committed Nov 1, 2023
1 parent 02994c5 commit 944fd03
Showing 1 changed file with 52 additions and 43 deletions.
95 changes: 52 additions & 43 deletions test/test_auto_registration.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,50 +49,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()

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
result = Mock()

if request.method == "PUT":
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):
Expand Down Expand Up @@ -120,7 +129,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
Expand Down Expand Up @@ -150,7 +159,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
Expand Down

0 comments on commit 944fd03

Please sign in to comment.