Merge branch 'master' into 4823-python-sdk-writable-collection-api
[arvados.git] / 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     SERVICE_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_service_errors(self):
23         self.check_get_message(self.SERVICE_ERRORS[:])
24
25     def test_get_message_without_service_errors(self):
26         self.check_get_message()
27
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())
32
33     def test_get_service_errors(self):
34         self.check_get_service_errors(self.SERVICE_ERRORS[:])
35
36     def test_get_service_errors_none(self):
37         self.check_get_service_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.service_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_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)
53
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.
64         last_index = -1
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