sdk/python/tests/test_retry_job_helpers.py

   1 from __future__ import absolute_import
   2 from builtins import object
   3 import mock
   4 import os
   5 import unittest
   6 import hashlib
   7 from . import run_test_server
   8 import json
   9 import arvados
  10 from . import arvados_testutil as tutil
  11 from apiclient import http as apiclient_http
  12
  13
  14 @tutil.skip_sleep
  15 class ApiClientRetryTestMixin(object):
  16
  17     TEST_UUID = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
  18     TEST_LOCATOR = 'd41d8cd98f00b204e9800998ecf8427e+0'
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         run_test_server.run()
  23
  24     def setUp(self):
  25         # Patch arvados.api() to return our mock API, so we can mock
  26         # its http requests.
  27         self.api_client = arvados.api('v1', cache=False)
  28         self.api_patch = mock.patch('arvados.api', return_value=self.api_client)
  29         self.api_patch.start()
  30
  31     def tearDown(self):
  32         self.api_patch.stop()
  33
  34     def run_method(self):
  35         raise NotImplementedError("test subclasses must define run_method")
  36
  37     def test_immediate_success(self):
  38         with tutil.mock_api_responses(self.api_client, '{}', [200]):
  39             self.run_method()
  40
  41     def test_immediate_failure(self):
  42         with tutil.mock_api_responses(self.api_client, '{}', [400]), self.assertRaises(self.DEFAULT_EXCEPTION):
  43             self.run_method()
  44
  45     def test_retry_then_success(self):
  46         with tutil.mock_api_responses(self.api_client, '{}', [500, 200]):
  47             self.run_method()
  48
  49     def test_error_after_default_retries_exhausted(self):
  50         with tutil.mock_api_responses(self.api_client, '{}', [500, 500, 500, 500, 500, 500, 200]), self.assertRaises(self.DEFAULT_EXCEPTION):
  51             self.run_method()
  52
  53     def test_no_retry_after_immediate_success(self):
  54         with tutil.mock_api_responses(self.api_client, '{}', [200, 400]):
  55             self.run_method()
  56
  57
  58 class CurrentJobTestCase(ApiClientRetryTestMixin, unittest.TestCase):
  59
  60     DEFAULT_EXCEPTION = arvados.errors.ApiError
  61
  62     def setUp(self):
  63         super(CurrentJobTestCase, self).setUp()
  64         os.environ['JOB_UUID'] = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
  65         os.environ['JOB_WORK'] = '.'
  66
  67     def tearDown(self):
  68         del os.environ['JOB_UUID']
  69         del os.environ['JOB_WORK']
  70         arvados._current_job = None
  71         super(CurrentJobTestCase, self).tearDown()
  72
  73     def run_method(self):
  74         arvados.current_job()
  75
  76
  77 class CurrentTaskTestCase(ApiClientRetryTestMixin, unittest.TestCase):
  78
  79     DEFAULT_EXCEPTION = arvados.errors.ApiError
  80
  81     def setUp(self):
  82         super(CurrentTaskTestCase, self).setUp()
  83         os.environ['TASK_UUID'] = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
  84         os.environ['TASK_WORK'] = '.'
  85
  86     def tearDown(self):
  87         del os.environ['TASK_UUID']
  88         del os.environ['TASK_WORK']
  89         arvados._current_task = None
  90         super(CurrentTaskTestCase, self).tearDown()
  91
  92     def run_method(self):
  93         arvados.current_task()
  94
  95
  96 class TaskSetOutputTestCase(CurrentTaskTestCase, unittest.TestCase):
  97
  98     DEFAULT_EXCEPTION = arvados.errors.ApiError
  99
 100     def tearDown(self):
 101         super(TaskSetOutputTestCase, self).tearDown()
 102         run_test_server.reset()
 103
 104     def run_method(self, locator=ApiClientRetryTestMixin.TEST_LOCATOR):
 105         arvados.task_set_output({'uuid':self.TEST_UUID},s=locator)