services/fuse/tests/test_retry.py

   1 import arvados
   2 import arvados_fuse.command
   3 import mock
   4 import os
   5 import run_test_server
   6 import tempfile
   7 import unittest
   8
   9 class KeepClientRetry(unittest.TestCase):
  10     origKeepClient = arvados.keep.KeepClient
  11
  12     def setUp(self):
  13         self.mnt = tempfile.mkdtemp()
  14         run_test_server.authorize_with('active')
  15
  16     def tearDown(self):
  17         os.rmdir(self.mnt)
  18
  19     @mock.patch('arvados_fuse.arvados.keep.KeepClient')
  20     def _test_retry(self, num_retries, argv, kc):
  21         kc.side_effect = lambda *args, **kw: self.origKeepClient(*args, **kw)
  22         with arvados_fuse.command.Mount(
  23                 arvados_fuse.command.ArgumentParser().parse_args(
  24                     argv+[self.mnt])):
  25             pass
  26         self.assertEqual(num_retries, kc.call_args[1].get('num_retries'))
  27
  28     def test_default_retry_3(self):
  29         self._test_retry(3, [])
  30
  31     def test_retry_2(self):
  32         self._test_retry(2, ['--retries=2'])
  33
  34     def test_no_retry(self):
  35         self._test_retry(0, ['--retries=0'])