|
9 | 9 | import unittest |
10 | 10 | import jwt |
11 | 11 | import botocore |
| 12 | +import requests |
| 13 | +import responses |
12 | 14 |
|
13 | 15 | from botocore import UNSIGNED |
14 | 16 | from botocore.client import Config as BotoConfig |
15 | 17 | from botocore.stub import Stubber, ANY |
16 | 18 |
|
17 | 19 | from staxapp.auth import StaxAuth |
| 20 | +from staxapp.config import Config |
18 | 21 | from staxapp.exceptions import InvalidCredentialsException |
19 | 22 |
|
20 | 23 |
|
@@ -197,6 +200,35 @@ def stub_cognito_creds(self, token: str): |
197 | 200 |
|
198 | 201 | self.cognito_stub.activate() |
199 | 202 |
|
| 203 | + @responses.activate |
| 204 | + def testSigV4Headers(self): |
| 205 | + """ |
| 206 | + Test sigv4 signed auth headers |
| 207 | + """ |
| 208 | + # Get signed auth headers |
| 209 | + sa = StaxAuth("ApiAuth") |
| 210 | + id_creds = { |
| 211 | + "Credentials": { |
| 212 | + "AccessKeyId": "ASIAX000000000000000", |
| 213 | + "SecretKey": "0000000000000000000000000000000000000000", |
| 214 | + "SessionToken": "a-totally-valid-JWT", |
| 215 | + "Expiration": datetime.datetime(2020, 1, 14, 11, 52, 26), |
| 216 | + } |
| 217 | + } |
| 218 | + auth = sa.sigv4_signed_auth_headers(id_creds) |
| 219 | + |
| 220 | + # Mock request |
| 221 | + response_dict = {"Status": "OK"} |
| 222 | + responses.add( |
| 223 | + responses.GET, |
| 224 | + f"{Config.api_base_url()}/auth", |
| 225 | + json=response_dict, |
| 226 | + status=200, |
| 227 | + ) |
| 228 | + response = requests.get(f"{Config.api_base_url()}/auth", auth=auth) |
| 229 | + self.assertEqual(response.json(), response_dict) |
| 230 | + self.assertIn("Authorization", response.request.headers) |
| 231 | + |
200 | 232 |
|
201 | 233 | if __name__ == "__main__": |
202 | 234 | unittest.main() |
0 commit comments