Repository navigation
Expand file tree
/
Copy pathtest_client.py
More file actions
151 lines (101 loc) · 5.11 KB
/
Copy pathtest_client.py
File metadata and controls
151 lines (101 loc) · 5.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
"""Tests that need no network: the ConvertAPI service is mocked with `responses`."""
import io
import json
from urllib.parse import parse_qs
import pytest
import responses
import convertapi
BASE = 'https://v2.convertapi.com/'
@pytest.fixture(autouse=True)
def credentials():
convertapi.api_credentials = 'test-token'
def conversion(*files):
return {
'ConversionCost': 1,
'Files': [{ 'FileName': name, 'FileSize': 4, 'Url': 'https://v2.convertapi.com/d/' + name } for name in files],
}
def form(call):
return { k: v[0] for k, v in parse_qs(call.request.body).items() }
def test_defaults():
assert convertapi.base_uri == BASE
assert convertapi.user_agent == 'ConvertAPI-Python/' + convertapi.__version__
def test_configuration():
convertapi.api_credentials = 'TEST'
assert convertapi.api_credentials == 'TEST'
@responses.activate
def test_user_sends_token_and_user_agent():
responses.get(BASE + 'user', json={ 'Active': True })
assert convertapi.user()['Active']
headers = responses.calls[0].request.headers
assert headers['Authorization'] == 'Bearer test-token'
assert headers['User-Agent'] == convertapi.user_agent
@responses.activate
def test_convert_url():
responses.post(BASE + 'convert/web/to/pdf', json=conversion('page.pdf'))
result = convertapi.convert('pdf', { 'Url': 'https://example.com' })
assert result.conversion_cost == 1
assert result.file.filename == 'page.pdf'
assert form(responses.calls[0]) == { 'Url': 'https://example.com', 'StoreFile': 'True' }
@responses.activate
def test_convert_with_format_and_timeout():
responses.post(BASE + 'convert/web/to/pdf', json=conversion('page.pdf'))
convertapi.convert('pdf', { 'Url': 'https://example.com/a.txt' }, 'web', 100)
assert form(responses.calls[0])['Timeout'] == '100'
assert responses.calls[0].request.req_kwargs['timeout'] == 100 + convertapi.conversion_timeout_delta
@responses.activate
def test_convert_file_uploads_it_first(tmp_path):
responses.post(BASE + 'upload', json={ 'FileId': 'abc', 'FileExt': 'docx' })
responses.post(BASE + 'convert/docx/to/pdf', json=conversion('test.pdf'))
responses.get('https://v2.convertapi.com/d/test.pdf', body=b'%PDF')
result = convertapi.convert('pdf', { 'File': 'examples/files/test.docx' })
saved = result.save_files(str(tmp_path))
upload = responses.calls[0].request
assert upload.headers['Content-Disposition'] == "attachment; filename*=UTF-8''test.docx"
assert form(responses.calls[1])['File'] == 'abc'
assert saved == [str(tmp_path / 'test.pdf')]
assert (tmp_path / 'test.pdf').read_bytes() == b'%PDF'
@responses.activate
def test_upload_io_and_download_io():
responses.post(BASE + 'upload', json={ 'FileId': 'abc', 'FileExt': 'txt' })
responses.post(BASE + 'convert/txt/to/pdf', json=conversion('test.pdf'))
responses.get('https://v2.convertapi.com/d/test.pdf', body=b'%PDF')
upload_io = convertapi.UploadIO(io.BytesIO(b'test'), 'test.txt')
result = convertapi.convert('pdf', { 'File': upload_io })
assert result.file.io.getvalue() == b'%PDF'
assert responses.calls[0].request.body == b'test'
@responses.activate
def test_chained_conversion_passes_result_urls():
responses.post(BASE + 'convert/any/to/zip', json=conversion('test.zip'))
previous = convertapi.result.Result(conversion('a.pdf', 'b.pdf'))
convertapi.convert('zip', { 'Files': previous.files })
sent = form(responses.calls[0])
assert sent['Files[0]'] == 'https://v2.convertapi.com/d/a.pdf'
assert sent['Files[1]'] == 'https://v2.convertapi.com/d/b.pdf'
@responses.activate
def test_api_error():
responses.post(BASE + 'convert/web/to/pdf', status=401, json={ 'Code': 4011, 'Message': 'Unauthorized.' })
with pytest.raises(convertapi.ApiError) as error:
convertapi.convert('pdf', { 'Url': 'https://example.com' })
assert error.value.code == 4011
assert str(error.value) == 'Unauthorized. Code: 4011.'
@responses.activate
def test_http_error_without_json_body():
responses.post(BASE + 'convert/web/to/pdf', status=502, body='Bad Gateway')
with pytest.raises(Exception) as error:
convertapi.convert('pdf', { 'Url': 'https://example.com' })
assert '502' in str(error.value)
@pytest.mark.parametrize('max_parallel_uploads', [1, 10])
@responses.activate
def test_convert_files_uploads_each(max_parallel_uploads):
# Each upload must carry the caller's configuration: in 2.0.0 parallel uploads ran in child processes,
# which lost api_credentials under the spawn and forkserver start methods (macOS, Python 3.14 on Linux).
convertapi.max_parallel_uploads = max_parallel_uploads
responses.post(BASE + 'upload', json={ 'FileId': 'abc', 'FileExt': 'docx' })
responses.post(BASE + 'convert/any/to/zip', json=conversion('test.zip'))
convertapi.convert('zip', { 'Files': ['examples/files/test.docx', 'examples/files/test.pdf'] })
uploads = [call.request for call in responses.calls if call.request.url == BASE + 'upload']
assert [r.headers['Authorization'] for r in uploads] == ['Bearer test-token'] * 2
assert sorted(r.headers['Content-Disposition'] for r in uploads) == [
"attachment; filename*=UTF-8''test.docx", "attachment; filename*=UTF-8''test.pdf"]
sent = form(responses.calls[-1])
assert sent['Files[0]'] == 'abc' and sent['Files[1]'] == 'abc'