convertapi-library-python/tests/test_convertapi.py at 9930a167ef445f77c96ed03bc4044cdd328346cc · ConvertAPI/convertapi-library-python · GitHub

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
import convertapi
import os
import io
import tempfile
import time
import requests

from . import utils
from nose.tools import *

class TestConvertapi(utils.TestCase):
	def setUp(self):
		convertapi.api_secret = os.environ['CONVERT_API_SECRET']
		convertapi.max_parallel_uploads = 10

	def test_defaults(self):
		eq_('https://v2.convertapi.com/', convertapi.base_uri)

	def test_configuration(self):
		convertapi.api_secret = 'TEST'
		eq_('TEST', convertapi.api_secret)

	def test_convert_file(self):
		result = convertapi.convert('pdf', { 'File': 'examples/files/test.docx' })
		assert result.save_files(tempfile.gettempdir())
		assert result.conversion_cost > 0

	def test_convert_file_no_parallelizm(self):
		convertapi.max_parallel_uploads = 1
		result = convertapi.convert('pdf', { 'File': 'examples/files/test.docx' })
		assert result.save_files(tempfile.gettempdir())

	def test_convert_file_alternative(self):
		result = convertapi.convert('pdf', { 'File': 'examples/files/test.docx', 'converter': 'openoffice' })
		assert result.save_files(tempfile.gettempdir())
		assert result.conversion_cost > 0

	def test_convert_file_url(self):
		result = convertapi.convert('pdf', { 'File': 'https://cdn.convertapi.com/cara/testfiles/document.docx?test=1' })
		assert result.conversion_cost > 0

	def test_convert_url(self):
		result = convertapi.convert('pdf', { 'Url': 'http://convertapi.com' })
		assert result.conversion_cost > 0

	def test_convert_url_with_timeout_and_format(self):
		result = convertapi.convert('pdf', { 'Url': 'https://www.w3.org/TR/PNG/iso_8859-1.txt' }, 'web', 100)
		assert result.conversion_cost > 0

	def test_upload_io(self):
		bytes_io = io.BytesIO(b'test')
		upload_io = convertapi.UploadIO(bytes_io, 'test.txt')
		result = convertapi.convert('pdf', { 'File': upload_io })
		assert result.conversion_cost > 0

	def test_download_io(self):
		result = convertapi.convert('pdf', { 'Url': 'https://www.w3.org/TR/PNG/iso_8859-1.txt' })
		assert len(result.file.io.getvalue()) > 0

	def test_zip_files(self):
		files = ['examples/files/test.docx', 'examples/files/test.docx']
		result = convertapi.convert('zip', { 'Files': files })
		assert result.conversion_cost > 0

	def test_chained_conversion(self):
		result = convertapi.convert('pdf', { 'File': 'examples/files/test.docx' })
		zip_result = convertapi.convert('zip', { 'Files': result.files })
		eq_('test.zip', zip_result.file.filename)

	@raises(convertapi.ApiError)
	def test_api_error(self):
		convertapi.api_secret = 'TEST'
		convertapi.convert('pdf', { 'Url': 'https://www.w3.org/TR/PNG/iso_8859-1.txt' })

	def test_user_info(self):
		user_info = convertapi.user()
		assert user_info['Active']


class TestAsyncConvertapi(utils.TestCase):
	retry_sleep_base_timeout = 0.1

	def setUp(self):
		convertapi.api_secret = os.environ['CONVERT_API_SECRET']
		convertapi.max_parallel_uploads = 10

	def test_async_conversion_and_polling(self):
		convert_result = convertapi.async_convert('pdf', { 'File': 'examples/files/test.docx' })
		assert convert_result.response['JobId']

		poll_result = self.get_poll_result(convert_result.response['JobId'])
		assert poll_result.save_files(tempfile.gettempdir())
		assert poll_result.conversion_cost > 0

	def test_polling_of_invalid_job_id(self):
		fake_job_id = 'rwvd6vtq58eurfwv5zw5h2ci2pgno1pn'
		try:
			convertapi.async_poll(fake_job_id)
		except requests.exceptions.HTTPError:
			pass
		else:
			raise AssertionError

	def test_polling_too_fast_and_getting_202_accepted(self):
		convert_result = convertapi.async_convert('pdf', { 'File': 'examples/files/test.docx' })
		try:
			convertapi.async_poll(convert_result.response['JobId'])
		except convertapi.AsyncConversionInProgress:
			pass
		else:
			raise AssertionError

	def get_poll_result(self, job_id, retry_count=5):
		try:
			result = convertapi.async_poll(job_id)
		except convertapi.AsyncConversionInProgress:
			if retry_count > 0:
				time.sleep((1 + 0.1) ** (5 - retry_count) - 1)
				return self.get_poll_result(job_id, retry_count=retry_count - 1)
			else:
				raise error
		else:
			return result