|
| 1 | +import copy |
| 2 | +import json |
| 3 | +import unittest |
| 4 | +import requests_mock |
| 5 | +from datetime import timedelta |
| 6 | +from transloadit.client import Transloadit |
| 7 | +from transloadit.request import Request |
| 8 | +from . import get_test_time, request_body_matcher |
| 9 | + |
| 10 | + |
| 11 | +class TestSignatureAuthentication(unittest.TestCase): |
| 12 | + def setUp(self): |
| 13 | + self.mock_client = MockClient("TRANSLOADIT_KEY", "TRANSLOADIT_SECRET") |
| 14 | + self.json_response = ( |
| 15 | + '{"ok": "ASSEMBLY_COMPLETED", "assembly_id": "abcdef45673"}' |
| 16 | + ) |
| 17 | + |
| 18 | + @requests_mock.Mocker() |
| 19 | + def test_fixed_signature(self, mock): |
| 20 | + # Test a request with a fixed timestamp in order to have reproducible results |
| 21 | + assembly = self.mock_client.new_assembly() |
| 22 | + assembly.add_step("import", "/http/import", |
| 23 | + options={"url": "https://demos.transloadit.com/inputs/chameleon.jpg"}) |
| 24 | + assembly.add_step("resize", "/image/resize", {"use:": "import", "width": 70, "height": 70}) |
| 25 | + |
| 26 | + url = f"{self.mock_client.service}/assemblies" |
| 27 | + mock.post( |
| 28 | + url, |
| 29 | + text=self.json_response, |
| 30 | + additional_matcher=request_body_matcher( |
| 31 | + "signature=sha384" |
| 32 | + "%3A46943b5542af95679940d94507865b20b36cb1808a7a9dc64c6255f26d1e921bd6574443b80b0dcd595769268f74273c"), |
| 33 | + ) |
| 34 | + |
| 35 | + assembly.create(resumable=False) |
| 36 | + |
| 37 | + |
| 38 | +class MockClient(Transloadit): |
| 39 | + """ |
| 40 | + Mock Class of the Transloadit Clients, which loads the modified MockRequest Class |
| 41 | + instead of the Standard Request class. |
| 42 | + """ |
| 43 | + |
| 44 | + def __init__(self, |
| 45 | + auth_key: str, |
| 46 | + auth_secret: str, |
| 47 | + service: str = "https://api2.transloadit.com", |
| 48 | + duration: int = 300): |
| 49 | + if not service.startswith(("http://", "https://")): |
| 50 | + service = "https://" + service |
| 51 | + |
| 52 | + self.service = service |
| 53 | + self.auth_key = auth_key |
| 54 | + self.auth_secret = auth_secret |
| 55 | + self.duration = duration |
| 56 | + self.request = MockRequest(self) |
| 57 | + |
| 58 | + |
| 59 | +class MockRequest(Request): |
| 60 | + """ |
| 61 | + Mock Request Class, which uses a fixed datetime for generating the signature. |
| 62 | + This is for having a reproducible value to test against. |
| 63 | + """ |
| 64 | + def _to_payload(self, data): |
| 65 | + data = copy.deepcopy(data or {}) |
| 66 | + expiry = timedelta(seconds=self.transloadit.duration) + get_test_time() |
| 67 | + data["auth"] = { |
| 68 | + "key": self.transloadit.auth_key, |
| 69 | + "expires": expiry.strftime("%Y/%m/%d %H:%M:%S+00:00"), |
| 70 | + } |
| 71 | + json_data = json.dumps(data) |
| 72 | + |
| 73 | + return {"params": json, "signature": self._sign_data(json_data)} |
0 commit comments