services/fuse/tests/test_token_expiry.py

   1 import arvados
   2 import apiclient
   3 import logging
   4 import mock
   5 import os
   6 import re
   7 import time
   8 import unittest
   9
  10 from .integration_test import IntegrationTest
  11
  12 logger = logging.getLogger('arvados.arv-mount')
  13
  14 class TokenExpiryTest(IntegrationTest):
  15     @mock.patch('arvados.keep.KeepClient.get')
  16     @IntegrationTest.mount(argv=['--mount-by-id', 'zzz'])
  17     def test_refresh_old_manifest(self, mocked_get):
  18         mocked_get.return_value = 'fake data'
  19
  20         self.api._rootDesc["blobSignatureTtl"] = 2
  21         old_exp = int(time.time()) + 86400*14
  22         self.pool_test(os.path.join(self.mnt, 'zzz'))
  23         want_exp = int(time.time()) + 86400*14
  24
  25         got_loc = mocked_get.call_args[0][0]
  26         got_exp = int(
  27             re.search(r'\+A[0-9a-f]+@([0-9a-f]+)', got_loc).group(1),
  28             16)
  29         self.assertGreaterEqual(
  30             got_exp, want_exp-2,
  31             msg='now+2w = {:x}, but fuse fetched locator {} (old_exp {:x})'.format(
  32                 want_exp, got_loc, old_exp))
  33         self.assertLessEqual(
  34             got_exp, want_exp,
  35             msg='server is not using the expected 2w TTL; test is ineffective')
  36
  37     @staticmethod
  38     def _test_refresh_old_manifest(self, zzz):
  39         uuid = 'zzzzz-4zz18-op4e2lbej01tcvu'
  40         fnm = 'zzzzz-8i9sb-0vsrcqi7whchuil.log.txt'
  41         os.listdir(os.path.join(zzz, uuid))
  42         time.sleep(3)
  43         with open(os.path.join(zzz, uuid, fnm)) as f:
  44             f.read()