#
# SPDX-License-Identifier: Apache-2.0
-from __future__ import absolute_import
-from builtins import object
-import mock
-import os
-import unittest
import hashlib
-from . import run_test_server
import json
-import arvados
-from . import arvados_testutil as tutil
+import os
+import unittest
+
from apiclient import http as apiclient_http
+from unittest import mock
+import arvados
+from . import run_test_server
+from . import arvados_testutil as tutil
@tutil.skip_sleep
class ApiClientRetryTestMixin(object):
def setUp(self):
# Patch arvados.api() to return our mock API, so we can mock
# its http requests.
- self.api_client = arvados.api('v1', cache=False)
+ self.api_client = arvados.api('v1', cache=False, num_retries=0)
self.api_patch = mock.patch('arvados.api', return_value=self.api_client)
self.api_patch.start()
def test_no_retry_after_immediate_success(self):
with tutil.mock_api_responses(self.api_client, '{}', [200, 400]):
self.run_method()
-
-
-class CurrentJobTestCase(ApiClientRetryTestMixin, unittest.TestCase):
-
- DEFAULT_EXCEPTION = arvados.errors.ApiError
-
- def setUp(self):
- super(CurrentJobTestCase, self).setUp()
- os.environ['JOB_UUID'] = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
- os.environ['JOB_WORK'] = '.'
-
- def tearDown(self):
- del os.environ['JOB_UUID']
- del os.environ['JOB_WORK']
- arvados._current_job = None
- super(CurrentJobTestCase, self).tearDown()
-
- def run_method(self):
- arvados.current_job()
-
-
-class CurrentTaskTestCase(ApiClientRetryTestMixin, unittest.TestCase):
-
- DEFAULT_EXCEPTION = arvados.errors.ApiError
-
- def setUp(self):
- super(CurrentTaskTestCase, self).setUp()
- os.environ['TASK_UUID'] = 'zzzzz-zzzzz-zzzzzzzzzzzzzzz'
- os.environ['TASK_WORK'] = '.'
-
- def tearDown(self):
- del os.environ['TASK_UUID']
- del os.environ['TASK_WORK']
- arvados._current_task = None
- super(CurrentTaskTestCase, self).tearDown()
-
- def run_method(self):
- arvados.current_task()
-
-
-class TaskSetOutputTestCase(CurrentTaskTestCase, unittest.TestCase):
-
- DEFAULT_EXCEPTION = arvados.errors.ApiError
-
- def tearDown(self):
- super(TaskSetOutputTestCase, self).tearDown()
- run_test_server.reset()
-
- def run_method(self, locator=ApiClientRetryTestMixin.TEST_LOCATOR):
- arvados.task_set_output({'uuid':self.TEST_UUID},s=locator)