services/fuse/tests/test_cache.py

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