6 import arvados.errors as arv_error
7 import arvados_testutil as tutil
9 class KeepRequestErrorTestCase(unittest.TestCase):
11 ('http://keep1.zzzzz.example.org/', IOError("test IOError")),
12 ('http://keep3.zzzzz.example.org/', MemoryError("test MemoryError")),
13 ('http://keep5.zzzzz.example.org/', tutil.fake_requests_response(
15 ('http://keep7.zzzzz.example.org/', IOError("second test IOError")),
18 def check_get_message(self, *args):
19 test_exc = arv_error.KeepRequestError("test message", *args)
20 self.assertEqual("test message", test_exc.message)
22 def test_get_message_with_service_errors(self):
23 self.check_get_message(self.SERVICE_ERRORS[:])
25 def test_get_message_without_service_errors(self):
26 self.check_get_message()
28 def check_get_service_errors(self, *args):
29 expected = dict(args[0]) if args else {}
30 test_exc = arv_error.KeepRequestError("test service exceptions", *args)
31 self.assertEqual(expected, test_exc.service_errors())
33 def test_get_service_errors(self):
34 self.check_get_service_errors(self.SERVICE_ERRORS[:])
36 def test_get_service_errors_none(self):
37 self.check_get_service_errors({})
39 def test_empty_exception(self):
40 test_exc = arv_error.KeepRequestError()
41 self.assertFalse(test_exc.message)
42 self.assertEqual({}, test_exc.service_errors())
44 def traceback_str(self, exc):
45 return traceback.format_exception_only(type(exc), exc)[-1]
47 def test_traceback_str_without_service_errors(self):
48 message = "test plain traceback string"
49 test_exc = arv_error.KeepRequestError(message)
50 exc_report = self.traceback_str(test_exc)
51 self.assertTrue(exc_report.startswith("KeepRequestError: "))
52 self.assertIn(message, exc_report)
54 def test_traceback_str_with_service_errors(self):
55 message = "test traceback shows Keep services"
56 test_exc = arv_error.KeepRequestError(message, self.SERVICE_ERRORS[:])
57 exc_report = self.traceback_str(test_exc)
58 self.assertTrue(exc_report.startswith("KeepRequestError: "))
59 for expect_substr in [message, "raised IOError", "raised MemoryError",
60 "test MemoryError", "second test IOError",
61 "responded with 500 Internal Server Error"]:
62 self.assertIn(expect_substr, exc_report)
63 # Assert the report maintains order of listed services.
65 for service_key, _ in self.SERVICE_ERRORS:
66 service_index = exc_report.find(service_key)
67 self.assertGreater(service_index, last_index)
68 last_index = service_index