15954: Merge branch 'master'
[arvados.git] / 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