|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | + |
| 3 | +import hashlib |
| 4 | +import unittest |
| 5 | + |
| 6 | +from freezegun import freeze_time |
| 7 | + |
| 8 | +from message_media_messages.api_helper import APIHelper |
| 9 | +from message_media_messages.configuration import Configuration |
| 10 | +from message_media_messages.http.auth.auth_manager import AuthManager |
| 11 | +from message_media_messages.http.requests_client import RequestsClient |
| 12 | +from message_media_messages.models.format_enum import FormatEnum |
| 13 | +from message_media_messages.models.message import Message |
| 14 | +from message_media_messages.models.send_messages_request import SendMessagesRequest |
| 15 | +from tests.test_util import TestUtil |
| 16 | + |
| 17 | + |
| 18 | +class AuthManagerTest(unittest.TestCase): |
| 19 | + |
| 20 | + def __init__(self, method_name: str = ...): |
| 21 | + super().__init__(method_name) |
| 22 | + self.content_hash = None |
| 23 | + self.query_url = None |
| 24 | + self.send_message_request = None |
| 25 | + |
| 26 | + def setUp(self): |
| 27 | + self.send_message_request = self.get_send_message_request() |
| 28 | + self.query_url = APIHelper.clean_url(Configuration.base_uri + '/v1/messages') |
| 29 | + self.content_hash = self.get_content_hash(self.send_message_request) |
| 30 | + self.content_md5_header = "x-Content-MD5: {}\n".format(self.content_hash) |
| 31 | + self.date_header = 'Wed, 08 Feb 2023 03:00:00 GMT' |
| 32 | + self.expected_algorithm = ' algorithm="hmac-sha1"' |
| 33 | + Configuration.hmac_auth_user_name = "some_user_name" |
| 34 | + self.expected_username = "hmac username=\"" + Configuration.hmac_auth_user_name + "\"" |
| 35 | + |
| 36 | + @freeze_time("2023-02-08 14:00:00") |
| 37 | + def test_post_request_hmac_authorization_header_values_are_appropriate(self): |
| 38 | + body = APIHelper.json_serialize(self.send_message_request) |
| 39 | + |
| 40 | + request = RequestsClient().post( |
| 41 | + query_url=self.query_url, |
| 42 | + parameters=body, |
| 43 | + headers={} |
| 44 | + ) |
| 45 | + |
| 46 | + AuthManager.apply_hmac_auth(request, self.query_url, body) |
| 47 | + |
| 48 | + expected_headers_property = ' headers="date x-Content-MD5 request-line"' |
| 49 | + expected_signature = TestUtil.create_signature(self.date_header, self.content_md5_header, |
| 50 | + self.query_url, 'POST', True) |
| 51 | + self.assert_auth_header(request, expected_headers_property, expected_signature) |
| 52 | + |
| 53 | + def test_post_request_content_md5_is_md5_of_request_body(self): |
| 54 | + body = APIHelper.json_serialize(self.send_message_request) |
| 55 | + |
| 56 | + request = RequestsClient().post( |
| 57 | + query_url=self.query_url, |
| 58 | + parameters=body, |
| 59 | + headers={} |
| 60 | + ) |
| 61 | + |
| 62 | + AuthManager.apply_hmac_auth(request, self.query_url, body) |
| 63 | + request_md5 = request.headers['x-Content-MD5'] |
| 64 | + assert self.content_hash == request_md5 |
| 65 | + |
| 66 | + @freeze_time("2023-02-08 14:00:00") |
| 67 | + def test_get_request_hmac_authorization_header_values_are_appropriate(self): |
| 68 | + request = RequestsClient().get( |
| 69 | + query_url=self.query_url, |
| 70 | + headers={} |
| 71 | + ) |
| 72 | + |
| 73 | + AuthManager.apply_hmac_auth(request, self.query_url) |
| 74 | + |
| 75 | + expected_headers_property = ' headers="date request-line"' |
| 76 | + expected_signature = TestUtil.create_signature(self.date_header, "", |
| 77 | + self.query_url, 'GET', True) |
| 78 | + self.assert_auth_header(request, expected_headers_property, expected_signature) |
| 79 | + |
| 80 | + def assert_auth_header(self, request, expected_headers_property, expected_signature): |
| 81 | + username, algorithm, headers, signature = request.headers['Authorization'].split(',') |
| 82 | + with self.subTest(): |
| 83 | + self.assertEqual(self.expected_username, username) |
| 84 | + self.assertEqual(expected_headers_property, headers) |
| 85 | + self.assertEqual(self.expected_algorithm, algorithm) |
| 86 | + self.assertEqual(expected_signature, signature) |
| 87 | + |
| 88 | + @staticmethod |
| 89 | + def get_send_message_request(): |
| 90 | + send_message_request = SendMessagesRequest() |
| 91 | + send_message_request.messages = [] |
| 92 | + send_message_request.messages.append(Message()) |
| 93 | + send_message_request.messages[0].content = 'My tests message' |
| 94 | + send_message_request.messages[0].destination_number = '+61491570006' |
| 95 | + send_message_request.messages[0].format = FormatEnum.SMS |
| 96 | + return send_message_request |
| 97 | + |
| 98 | + @staticmethod |
| 99 | + def get_content_hash(request): |
| 100 | + m = hashlib.md5() |
| 101 | + m.update(bytes(APIHelper.json_serialize(request), 'utf-8')) |
| 102 | + content_hash = m.hexdigest() |
| 103 | + return content_hash |
0 commit comments