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
|
import unittest
from urllib3 import HTTPResponse
from influxdb_client.client.exceptions import InfluxDBError
class TestInfluxDBError(unittest.TestCase):
def test_response(self):
response = HTTPResponse()
self.assertEqual(response, InfluxDBError(response=response).response)
def test_message(self):
response = HTTPResponse()
response.headers.add('X-Platform-Error-Code', 'too many requests 1')
self.assertEqual("too many requests 1", str(InfluxDBError(response=response)))
response = HTTPResponse()
response.headers.add('X-Influx-Error', 'too many requests 2')
self.assertEqual("too many requests 2", str(InfluxDBError(response=response)))
response = HTTPResponse()
response.headers.add('X-InfluxDb-Error', 'too many requests 3')
self.assertEqual("too many requests 3", str(InfluxDBError(response=response)))
response = HTTPResponse(body='{"code":"too many requests","message":"org 04014de4ed590000 has exceeded limited_write plan limit"}')
response.headers.add('X-InfluxDb-Error', 'error 3')
self.assertEqual("org 04014de4ed590000 has exceeded limited_write plan limit", str(InfluxDBError(response=response)))
response = HTTPResponse(body='org 04014de4ed590000 has exceeded limited_write plan limit')
response.headers.add('X-InfluxDb-Error', 'error 3')
self.assertEqual("org 04014de4ed590000 has exceeded limited_write plan limit", str(InfluxDBError(response=response)))
response = HTTPResponse(reason='too many requests 4')
self.assertEqual("too many requests 4", str(InfluxDBError(response=response)))
def test_message_get_retry_after(self):
response = HTTPResponse(reason="too many requests")
response.headers.add('Retry-After', '63')
influx_db_error = InfluxDBError(response=response)
self.assertEqual("too many requests", str(influx_db_error))
self.assertEqual("63", influx_db_error.retry_after)
influx_db_error = InfluxDBError(response=HTTPResponse(reason="too many requests"))
self.assertEqual("too many requests", str(influx_db_error))
self.assertEqual(None, influx_db_error.retry_after)
def test_no_response(self):
influx_db_error = InfluxDBError(response=None)
self.assertEqual("no response", str(influx_db_error))
self.assertIsNone(influx_db_error.response)
self.assertIsNone(influx_db_error.retry_after)
|
