services/fuse/tests/test_cache.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 arvados
   7 import arvados.collection
   8 import arvados_fuse
   9 import arvados_fuse.command
  10 import json
  11 import logging
  12 import os
  13 import tempfile
  14 import unittest
  15
  16 from .integration_test import IntegrationTest
  17 from .mount_test_base import MountTestBase
  18
  19 class CacheTest(IntegrationTest):
  20     mnt_args = ["--by-id", "--directory-cache=0"]
  21
  22     @IntegrationTest.mount(argv=mnt_args)
  23     def test_cache_spill(self):
  24         pdh = []
  25         for i in range(0, 8):
  26             cw = arvados.collection.Collection()
  27             f = cw.open("blurg%i" % i, "w")
  28             f.write("bloop%i" % i)
  29
  30             cw.mkdirs("dir%i" % i)
  31             f = cw.open("dir%i/blurg" % i, "w")
  32             f.write("dirbloop%i" % i)
  33
  34             cw.save_new()
  35             pdh.append(cw.portable_data_hash())
  36         self.pool_test(self.mnt, pdh)
  37
  38     @staticmethod
  39     def _test_cache_spill(self, mnt, pdh):
  40         for i,v in enumerate(pdh):
  41             j = os.path.join(mnt, "by_id", v, "blurg%i" % i)
  42             self.assertTrue(os.path.exists(j))
  43             j = os.path.join(mnt, "by_id", v, "dir%i/blurg" % i)
  44             self.assertTrue(os.path.exists(j))
  45
  46         for i,v in enumerate(pdh):
  47             j = os.path.join(mnt, "by_id", v, "blurg%i" % i)
  48             self.assertTrue(os.path.exists(j))
  49             j = os.path.join(mnt, "by_id", v, "dir%i/blurg" % i)
  50             self.assertTrue(os.path.exists(j))