sdk/python/tests/test_errors.py

   1 #!/usr/bin/env python
   2
   3 import traceback
   4 import unittest
   5
   6 import arvados.errors as arv_error
   7 import arvados_testutil as tutil
   8
   9 class KeepRequestErrorTestCase(unittest.TestCase):
  10     REQUEST_ERRORS = [
  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(
  14                 500, "test 500")),
  15         ('http://keep7.zzzzz.example.org/', IOError("second test IOError")),
  16         ]
  17
  18     def check_get_message(self, *args):
  19         test_exc = arv_error.KeepRequestError("test message", *args)
  20         self.assertEqual("test message", test_exc.message)
  21
  22     def test_get_message_with_request_errors(self):
  23         self.check_get_message(self.REQUEST_ERRORS[:])
  24
  25     def test_get_message_without_request_errors(self):
  26         self.check_get_message()
  27
  28     def check_get_request_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.request_errors())
  32
  33     def test_get_request_errors(self):
  34         self.check_get_request_errors(self.REQUEST_ERRORS[:])
  35
  36     def test_get_request_errors_none(self):
  37         self.check_get_request_errors({})
  38
  39     def test_empty_exception(self):
  40         test_exc = arv_error.KeepRequestError()
  41         self.assertFalse(test_exc.message)
  42         self.assertEqual({}, test_exc.request_errors())
  43
  44     def traceback_str(self, exc):
  45         return traceback.format_exception_only(type(exc), exc)[-1]
  46
  47     def test_traceback_str_without_request_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)
  53
  54     def test_traceback_str_with_request_errors(self):
  55         message = "test traceback shows Keep services"
  56         test_exc = arv_error.KeepRequestError(message, self.REQUEST_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.
  64         last_index = -1
  65         for service_key, _ in self.REQUEST_ERRORS:
  66             service_index = exc_report.find(service_key)
  67             self.assertGreater(service_index, last_index)
  68             last_index = service_index