services/fuse/tests/test_token_expiry.py

   1 import apiclient
   2 import arvados
   3 import arvados_fuse
   4 import logging
   5 import mock
   6 import multiprocessing
   7 import os
   8 import re
   9 import sys
  10 import time
  11 import unittest
  12
  13 from .integration_test import IntegrationTest
  14
  15 logger = logging.getLogger('arvados.arv-mount')
  16
  17 class TokenExpiryTest(IntegrationTest):
  18     def __init__(self, *args, **kwargs):
  19         super(TokenExpiryTest, self).__init__(*args, **kwargs)
  20         self.test_start_time = time.time()
  21         self.time_now = int(time.time())+1
  22
  23     def fake_time(self):
  24         self.time_now += 1
  25         return self.time_now
  26
  27     orig_open = arvados_fuse.Operations.open
  28     def fake_open(self, operations, *args, **kwargs):
  29         self.time_now += 86400*13
  30         logger.debug('opening file at time=%f', self.time_now)
  31         return self.orig_open(operations, *args, **kwargs)
  32
  33     @mock.patch.object(arvados_fuse.Operations, 'open', autospec=True)
  34     @mock.patch('time.time')
  35     @mock.patch('arvados.keep.KeepClient.get')
  36     @IntegrationTest.mount(argv=['--mount-by-id', 'zzz'])
  37     def test_refresh_old_manifest(self, mocked_get, mocked_time, mocked_open):
  38         mocked_get.return_value = 'fake data'
  39         mocked_time.side_effect = self.fake_time
  40         mocked_open.side_effect = self.fake_open
  41
  42         with mock.patch.object(self.mount.api, 'collections', wraps=self.mount.api.collections) as mocked_collections:
  43             mocked_collections.return_value = mocked_collections()
  44             with mock.patch.object(self.mount.api.collections(), 'get', wraps=self.mount.api.collections().get) as mocked_get:
  45                 self.pool_test(os.path.join(self.mnt, 'zzz'))
  46
  47         self.assertEqual(3, mocked_open.call_count)
  48         self.assertEqual(
  49             4, mocked_get.call_count,
  50             'Not enough calls to collections().get(): expected 4, got {!r}'.format(
  51                 mocked_get.mock_calls))
  52
  53     @staticmethod
  54     def _test_refresh_old_manifest(self, zzz):
  55         uuid = 'zzzzz-4zz18-op4e2lbej01tcvu'
  56         fnm = 'zzzzz-8i9sb-0vsrcqi7whchuil.log.txt'
  57         os.listdir(os.path.join(zzz, uuid))
  58         for _ in range(3):
  59             with open(os.path.join(zzz, uuid, fnm)) as f:
  60                 f.read()