services/fuse/tests/test_retry.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 from __future__ import absolute_import
   6 from future import standard_library
   7 standard_library.install_aliases()
   8 import arvados
   9 import arvados_fuse.command
  10 import json
  11 import mock
  12 import os
  13 import pycurl
  14 import queue
  15 from . import run_test_server
  16 import tempfile
  17 import unittest
  18
  19 from .integration_test import IntegrationTest
  20
  21
  22 class KeepClientRetry(unittest.TestCase):
  23     origKeepClient = arvados.keep.KeepClient
  24
  25     def setUp(self):
  26         self.mnt = tempfile.mkdtemp()
  27         run_test_server.authorize_with('active')
  28
  29     def tearDown(self):
  30         os.rmdir(self.mnt)
  31
  32     @mock.patch('arvados_fuse.arvados.keep.KeepClient')
  33     def _test_retry(self, num_retries, argv, kc):
  34         kc.side_effect = lambda *args, **kw: self.origKeepClient(*args, **kw)
  35         with arvados_fuse.command.Mount(
  36                 arvados_fuse.command.ArgumentParser().parse_args(
  37                     argv+[self.mnt])):
  38             pass
  39         self.assertEqual(num_retries, kc.call_args[1].get('num_retries'))
  40
  41     def test_default_retry_10(self):
  42         self._test_retry(10, [])
  43
  44     def test_retry_2(self):
  45         self._test_retry(2, ['--retries=2'])
  46
  47     def test_no_retry(self):
  48         self._test_retry(0, ['--retries=0'])
  49
  50 class RetryPUT(IntegrationTest):
  51     @mock.patch('time.sleep')
  52     @IntegrationTest.mount(argv=['--read-write', '--mount-tmp=zzz'])
  53     def test_retry_write(self, sleep):
  54         mockedCurl = mock.Mock(spec=pycurl.Curl(), wraps=pycurl.Curl())
  55         mockedCurl.perform.side_effect = Exception('mock error (ok)')
  56         q = queue.Queue()
  57         q.put(mockedCurl)
  58         q.put(pycurl.Curl())
  59         q.put(pycurl.Curl())
  60         with mock.patch('arvados.keep.KeepClient.KeepService._get_user_agent', side_effect=q.get_nowait):
  61             self.pool_test(os.path.join(self.mnt, 'zzz'))
  62             self.assertTrue(mockedCurl.perform.called)
  63     @staticmethod
  64     def _test_retry_write(self, tmp):
  65         with open(os.path.join(tmp, 'foo'), 'w') as f:
  66             f.write('foo')
  67         json.load(open(os.path.join(tmp, '.arvados#collection')))