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 os
   8 import unittest
   9 import hashlib
  10 from . import run_test_server
  11 import json
  12 import arvados
  13 from . import arvados_testutil as tutil
  14 from apiclient import http as apiclient_http
  15
  16 from unittest import mock
  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()