()
| 201 | |
| 202 | |
| 203 | def test_upload(): |
| 204 | expected_s3_headers = { |
| 205 | 'content-length': '7557', |
| 206 | 'x-archive-queue-derive': '1', |
| 207 | 'x-archive-size-hint': '7557', |
| 208 | 'x-archive-auto-make-bucket': '1', |
| 209 | 'authorization': 'LOW test_access:test_secret', |
| 210 | } |
| 211 | with IaRequestsMock(assert_all_requests_are_fired=False) as rsps: |
| 212 | rsps.add(responses.PUT, re.compile(r'.*s3.us.archive.org/.*'), |
| 213 | adding_headers=expected_s3_headers) |
| 214 | rsps.add_metadata_mock('nasa') |
| 215 | rsps.add(responses.GET, f'{PROTOCOL}//archive.org/metadata/nasa', |
| 216 | body='{}') |
| 217 | _responses = upload('nasa', NASA_METADATA_PATH, |
| 218 | access_key='test_access', |
| 219 | secret_key='test_secret') |
| 220 | for response in _responses: |
| 221 | req = response.request |
| 222 | headers = {k.lower(): str(v) for k, v in req.headers.items()} |
| 223 | assert 'user-agent' in headers |
| 224 | del headers['accept'] |
| 225 | del headers['accept-encoding'] |
| 226 | del headers['connection'] |
| 227 | del headers['user-agent'] |
| 228 | assert headers == expected_s3_headers |
| 229 | assert req.url == f'{PROTOCOL}//s3.us.archive.org/nasa/nasa.json' |
| 230 | |
| 231 | |
| 232 | def test_upload_validate_identifier(): |
nothing calls this directly
no test coverage detected