Merge branch '20680-default-config-updates' refs #20680
[arvados.git] / sdk / python / tests / test_retry_job_helpers.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 from builtins import object
7 import mock
8 import os
9 import unittest
10 import hashlib
11 from . import run_test_server
12 import json
13 import arvados
14 from . import arvados_testutil as tutil
15 from apiclient import http as apiclient_http
16
17
18 @tutil.skip_sleep
19 class ApiClientRetryTestMixin(object):
20
21     TEST_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
22     TEST_LOCATOR = 'd41d8cd98f00b204e9800998ecf8427e+0'
23
24     @classmethod
25     def setUpClass(cls):
26         run_test_server.run()
27
28     def setUp(self):
29         # Patch arvados.api() to return our mock API, so we can mock
30         # its http requests.
31         self.api_client = arvados.api('v1', cache=False, num_retries=0)
32         self.api_patch = mock.patch('arvados.api', return_value=self.api_client)
33         self.api_patch.start()
34
35     def tearDown(self):
36         self.api_patch.stop()
37
38     def run_method(self):
39         raise NotImplementedError("test subclasses must define run_method")
40
41     def test_immediate_success(self):
42         with tutil.mock_api_responses(self.api_client, '{}', [200]):
43             self.run_method()
44
45     def test_immediate_failure(self):
46         with tutil.mock_api_responses(self.api_client, '{}', [400]), self.assertRaises(self.DEFAULT_EXCEPTION):
47             self.run_method()
48
49     def test_retry_then_success(self):
50         with tutil.mock_api_responses(self.api_client, '{}', [500, 200]):
51             self.run_method()
52
53     def test_error_after_default_retries_exhausted(self):
54         with tutil.mock_api_responses(self.api_client, '{}', [500, 500, 500, 500, 500, 500, 200]), self.assertRaises(self.DEFAULT_EXCEPTION):
55             self.run_method()
56
57     def test_no_retry_after_immediate_success(self):
58         with tutil.mock_api_responses(self.api_client, '{}', [200, 400]):
59             self.run_method()
60
61
62 class CurrentJobTestCase(ApiClientRetryTestMixin, unittest.TestCase):
63
64     DEFAULT_EXCEPTION = arvados.errors.ApiError
65
66     def setUp(self):
67         super(CurrentJobTestCase, self).setUp()
68         os.environ['JOB_UUID'] = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
69         os.environ['JOB_WORK'] = '.'
70
71     def tearDown(self):
72         del os.environ['JOB_UUID']
73         del os.environ['JOB_WORK']
74         arvados._current_job = None
75         super(CurrentJobTestCase, self).tearDown()
76
77     def run_method(self):
78         arvados.current_job()