services/fuse/tests/test_retry.py

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