sdk/python/tests/test_retry_job_helpers.py

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