sdk/python/tests/test_errors.py

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