1 from __future__ import absolute_import
5 import arvados.errors as arv_error
6 from . import arvados_testutil as tutil
8 class KeepRequestErrorTestCase(unittest.TestCase):
10 ('http://keep1.zzzzz.example.org/', IOError("test IOError")),
11 ('http://keep3.zzzzz.example.org/', MemoryError("test MemoryError")),
12 ('http://keep5.zzzzz.example.org/',
13 arv_error.HttpError(500, "Internal Server Error")),
14 ('http://keep7.zzzzz.example.org/', IOError("second test IOError")),
17 def check_get_message(self, *args):
18 test_exc = arv_error.KeepRequestError("test message", *args)
19 self.assertEqual("test message", test_exc.message)
21 def test_get_message_with_request_errors(self):
22 self.check_get_message(self.REQUEST_ERRORS[:])
24 def test_get_message_without_request_errors(self):
25 self.check_get_message()
27 def check_get_request_errors(self, *args):
28 expected = dict(args[0]) if args else {}
29 test_exc = arv_error.KeepRequestError("test service exceptions", *args)
30 self.assertEqual(expected, test_exc.request_errors())
32 def test_get_request_errors(self):
33 self.check_get_request_errors(self.REQUEST_ERRORS[:])
35 def test_get_request_errors_none(self):
36 self.check_get_request_errors({})
38 def test_empty_exception(self):
39 test_exc = arv_error.KeepRequestError()
40 self.assertFalse(test_exc.message)
41 self.assertEqual({}, test_exc.request_errors())
43 def traceback_str(self, exc):
44 return traceback.format_exception_only(type(exc), exc)[-1]
46 def test_traceback_str_without_request_errors(self):
47 message = "test plain traceback string"
48 test_exc = arv_error.KeepRequestError(message)
49 exc_report = self.traceback_str(test_exc)
50 self.assertRegex(exc_report, r"^(arvados\.errors\.)?KeepRequestError: ")
51 self.assertIn(message, exc_report)
53 def test_traceback_str_with_request_errors(self):
54 message = "test traceback shows Keep services"
55 test_exc = arv_error.KeepRequestError(message, self.REQUEST_ERRORS[:])
56 exc_report = self.traceback_str(test_exc)
57 self.assertRegex(exc_report, r"^(arvados\.errors\.)?KeepRequestError: ")
58 self.assertIn(message, exc_report)
60 r"raised (IOError|OSError)", # IOError in Python2, OSError in Python3
61 r"raised MemoryError",
63 r"second test IOError",
64 r"responded with 500 Internal Server Error"]:
65 self.assertRegex(exc_report, expect_re)
66 # Assert the report maintains order of listed services.
68 for service_key, _ in self.REQUEST_ERRORS:
69 service_index = exc_report.find(service_key)
70 self.assertGreater(service_index, last_index)
71 last_index = service_index