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()