#
# SPDX-License-Identifier: AGPL-3.0
-from __future__ import absolute_import
-from future.utils import viewitems
-from builtins import str
-from builtins import object
-from pathlib import Path
-from six import assertRegex
import errno
import json
import llfuse
import time
import unittest
import tempfile
-import parameterized
+from pathlib import Path
from unittest import mock
import arvados
import arvados_fuse as fuse
+import parameterized
+
from arvados_fuse import fusedir
-from . import run_test_server
+from . import run_test_server
from .integration_test import IntegrationTest
from .mount_test_base import MountTestBase
from .test_tmp_collection import storage_classes_desired
logger = logging.getLogger('arvados.arv-mount')
-
class AssertWithTimeout(object):
"""Allow some time for an assertion to pass."""
'dir2/dir3/thing7.txt': 'data 7',
'dir2/dir3/thing8.txt': 'data 8'}
- for k, v in viewitems(files):
+ for k, v in files.items():
with open(os.path.join(self.mounttmp, k), 'rb') as f:
self.assertEqual(v, f.read().decode())
files = {}
files[os.path.join(self.mounttmp, self.testcollection, 'thing1.txt')] = 'data 1'
- for k, v in viewitems(files):
+ for k, v in files.items():
with open(os.path.join(self.mounttmp, k), 'rb') as f:
self.assertEqual(v, f.read().decode())
'anonymously_accessible_project']
found_in = 0
found_not_in = 0
- for name, item in viewitems(run_test_server.fixture('collections')):
+ for name, item in run_test_server.fixture('collections').items():
if 'name' not in item:
pass
elif item['owner_uuid'] == public_project['uuid']:
self.assertEqual(["file1.txt"], d1)
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\. d41d8cd98f00b204e9800998ecf8427e\+0\+A\S+ 0:0:file1\.txt$')
self.assertEqual(12, self.operations.read_counter.get())
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
self.pool.apply(fuseUpdateFileTestHelper, (self.mounttmp,))
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\. daaef200ebb921e011e3ae922dd3266b\+11\+A\S+ 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:11:file1\.txt 22:1:file1\.txt$')
self.pool.apply(fuseMkdirTestHelper, (self.mounttmp,))
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
# Starting manifest
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
self.pool.apply(fuseRmTestHelperDeleteFile, (self.mounttmp,))
# Empty directories are represented by an empty file named "."
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'./testdir d41d8cd98f00b204e9800998ecf8427e\+0\+A\S+ 0:0:\\056\n')
self.pool.apply(fuseRmTestHelperRmdir, (self.mounttmp,))
# Starting manifest
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
self.pool.apply(fuseMvFileTestHelperMoveFile, (self.mounttmp,))
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt\n\./testdir d41d8cd98f00b204e9800998ecf8427e\+0\+A\S+ 0:0:\\056\n')
# Starting manifest
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
d1 = llfuse.listdir(os.path.join(self.mounttmp))
self.assertEqual(["file1.txt"], d1)
collection2 = self.api.collections().get(uuid=collection.manifest_locator()).execute()
- assertRegex(self, collection2["manifest_text"],
+ self.assertRegex(collection2["manifest_text"],
r'\./testdir2 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$')
with open(os.path.join(mounttmp, "file1.txt"), "r") as f:
self.assertEqual(f.read(), "bar")
- assertRegex(self, d1[1],
+ self.assertRegex(d1[1],
r'file1\.txt~\d\d\d\d\d\d\d\d-\d\d\d\d\d\d~conflict~')
with open(os.path.join(mounttmp, d1[1]), "r") as f:
collection1.update()
collection2.update()
- assertRegex(self, collection1.manifest_text(), r"\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
+ self.assertRegex(collection1.manifest_text(), r"\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
self.assertEqual(collection2.manifest_text(), "")
self.pool.apply(fuseMvFileBetweenCollectionsTest2, (self.mounttmp,
collection2.update()
self.assertEqual(collection1.manifest_text(), "")
- assertRegex(self, collection2.manifest_text(), r"\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file2\.txt$")
+ self.assertRegex(collection2.manifest_text(), r"\. 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file2\.txt$")
collection1.stop_threads()
collection2.stop_threads()
collection1.update()
collection2.update()
- assertRegex(self, collection1.manifest_text(), r"\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
+ self.assertRegex(collection1.manifest_text(), r"\./testdir 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
self.assertEqual(collection2.manifest_text(), "")
self.pool.apply(fuseMvDirBetweenCollectionsTest2, (self.mounttmp,
collection2.update()
self.assertEqual(collection1.manifest_text(), "")
- assertRegex(self, collection2.manifest_text(), r"\./testdir2 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
+ self.assertRegex(collection2.manifest_text(), r"\./testdir2 86fb269d190d2c85f6e0468ceca42a20\+12\+A\S+ 0:12:file1\.txt$")
collection1.stop_threads()
collection2.stop_threads()
files = {}
files[os.path.join(self.mounttmp, self.testcollection, 'thing1.txt')] = 'data 1'
- for k, v in viewitems(files):
+ for k, v in files.items():
with open(os.path.join(self.mounttmp, k), 'rb') as f:
self.assertEqual(v, f.read().decode())