services/fuse/tests/test_token_expiry.py

   1 # Copyright (C) The Arvados Authors. All rights reserved.
   2 #
   3 # SPDX-License-Identifier: AGPL-3.0
   4
   5 from builtins import range
   6 import apiclient
   7 import arvados
   8 import arvados_fuse
   9 import logging
  10 import multiprocessing
  11 import os
  12 import re
  13 import sys
  14 import time
  15 import unittest
  16
  17 from unittest import mock
  18
  19 from .integration_test import IntegrationTest
  20
  21 logger = logging.getLogger('arvados.arv-mount')
  22
  23 class TokenExpiryTest(IntegrationTest):
  24     def setUp(self):
  25         super(TokenExpiryTest, self).setUp()
  26         self.test_start_time = time.time()
  27         self.time_now = int(time.time())+1
  28
  29     def fake_time(self):
  30         self.time_now += 1
  31         return self.time_now
  32
  33     orig_open = arvados_fuse.Operations.open
  34     def fake_open(self, operations, *args, **kwargs):
  35         self.time_now += 86400*13
  36         logger.debug('opening file at time=%f', self.time_now)
  37         return TokenExpiryTest.orig_open(operations, *args, **kwargs)
  38
  39     @mock.patch.object(arvados_fuse.Operations, 'open', autospec=True)
  40     @mock.patch.object(time, 'time', return_value=0)
  41     @mock.patch('arvados.keep.KeepClient.get')
  42     @IntegrationTest.mount(argv=['--mount-by-id', 'zzz'])
  43     def test_refresh_old_manifest(self, mocked_get, mocked_time, mocked_open):
  44         # This test (and associated behavior) is still not strong
  45         # enough. We should ensure old tokens are never used even if
  46         # blobSignatureTtl seconds elapse between open() and
  47         # read(). See https://dev.arvados.org/issues/10008
  48
  49         mocked_get.return_value = b'fake data'
  50         mocked_time.side_effect = self.fake_time
  51         mocked_open.side_effect = self.fake_open
  52
  53         with mock.patch.object(self.mount.api, 'collections', wraps=self.mount.api.collections) as mocked_collections:
  54             mocked_collections.return_value = mocked_collections()
  55             with mock.patch.object(self.mount.api.collections(), 'get', wraps=self.mount.api.collections().get) as mocked_get:
  56                 self.pool_test(os.path.join(self.mnt, 'zzz'))
  57
  58         # open() several times here to make sure we don't reach our
  59         # quota of mocked_get.call_count dishonestly (e.g., the first
  60         # open causes 5 mocked_get, and the rest cause none).
  61         self.assertEqual(8, mocked_open.call_count)
  62         self.assertGreaterEqual(
  63             mocked_get.call_count, 8,
  64             'Not enough calls to collections().get(): expected 8, got {!r}'.format(
  65                 mocked_get.mock_calls))
  66
  67     @staticmethod
  68     def _test_refresh_old_manifest(self, zzz):
  69         uuid = 'zzzzz-4zz18-op4e2lbej01tcvu'
  70         fnm = 'zzzzz-8i9sb-0vsrcqi7whchuil.log.txt'
  71         os.listdir(os.path.join(zzz, uuid))
  72         for _ in range(8):
  73             with open(os.path.join(zzz, uuid, fnm)) as f:
  74                 f.read()