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