1 # Copyright (C) The Arvados Authors. All rights reserved.
3 # SPDX-License-Identifier: AGPL-3.0
11 from pathlib import Path
13 from parameterized import parameterized
15 from arvados_fuse import fusedir
17 from .integration_test import IntegrationTest
18 from .mount_test_base import MountTestBase
19 from .run_test_server import fixture
21 _COLLECTIONS = fixture('collections')
22 _GROUPS = fixture('groups')
23 _LINKS = fixture('links')
24 _USERS = fixture('users')
26 class DirectoryFiltersTestCase(MountTestBase):
27 DEFAULT_ROOT_KWARGS = {
28 'enable_write': False,
30 ['collections.name', 'like', 'zzzzz-4zz18-%'],
31 # This matches both "A Project" (which we use as the test root)
32 # and "A Subproject" (which we assert is found under it).
33 ['groups.name', 'like', 'A %roject'],
36 EXPECTED_PATHS = frozenset([
37 _COLLECTIONS['foo_collection_in_aproject']['name'],
38 _GROUPS['asubproject']['name'],
40 CHECKED_PATHS = EXPECTED_PATHS.union([
41 _COLLECTIONS['collection_to_move_around_in_aproject']['name'],
42 _GROUPS['subproject_in_active_user_home_project_to_test_unique_key_violation']['name'],
45 @parameterized.expand([
46 (fusedir.MagicDirectory, {}, _GROUPS['aproject']['uuid']),
47 (fusedir.ProjectDirectory, {'project_object': _GROUPS['aproject']}, '.'),
48 (fusedir.SharedDirectory, {'exclude': None}, Path(
49 '{first_name} {last_name}'.format_map(_USERS['active']),
50 _GROUPS['aproject']['name'],
53 def test_filtered_path_exists(self, root_class, root_kwargs, subdir):
54 root_kwargs = collections.ChainMap(root_kwargs, self.DEFAULT_ROOT_KWARGS)
55 self.make_mount(root_class, **root_kwargs)
56 dir_path = Path(self.mounttmp, subdir)
59 for basename in self.CHECKED_PATHS
60 if (dir_path / basename).exists()
65 "mount existence checks did not match expected results",
68 @parameterized.expand([
69 (fusedir.MagicDirectory, {}, _GROUPS['aproject']['uuid']),
70 (fusedir.ProjectDirectory, {'project_object': _GROUPS['aproject']}, '.'),
71 (fusedir.SharedDirectory, {'exclude': None}, Path(
72 '{first_name} {last_name}'.format_map(_USERS['active']),
73 _GROUPS['aproject']['name'],
76 def test_filtered_path_listing(self, root_class, root_kwargs, subdir):
77 root_kwargs = collections.ChainMap(root_kwargs, self.DEFAULT_ROOT_KWARGS)
78 self.make_mount(root_class, **root_kwargs)
79 actual = frozenset(path.name for path in Path(self.mounttmp, subdir).iterdir())
81 actual & self.EXPECTED_PATHS,
83 "mount listing did not include minimum matches",
88 if not (name.startswith('zzzzz-4zz18-') or name.endswith('roject'))
92 "mount listing included results outside filters",
96 class TagFiltersTestCase(MountTestBase):
97 COLL_UUID = _COLLECTIONS['foo_collection_in_aproject']['uuid']
98 TAG_NAME = _LINKS['foo_collection_tag']['name']
100 @parameterized.expand([
104 def test_tag_directory_filters(self, op):
106 fusedir.TagDirectory,
109 ['links.head_uuid', op, self.COLL_UUID],
113 checked_path = Path(self.mounttmp, self.COLL_UUID)
114 self.assertEqual(checked_path.exists(), op == '=')
116 @parameterized.expand(itertools.product(
120 def test_tags_directory_filters(self, coll_op, link_op):
122 fusedir.TagsDirectory,
125 ['links.head_uuid', coll_op, [self.COLL_UUID]],
126 ['links.name', link_op, self.TAG_NAME],
130 filtered_path = Path(self.mounttmp, self.TAG_NAME)
131 elif coll_op == 'not in':
132 # As of 2024-02-09, foo tag only applies to the single collection.
133 # If you filter it out via head_uuid, then it disappears completely
134 # from the TagsDirectory. Hence we set that tag directory as
135 # filtered_path. If any of this changes in the future,
136 # it would be fine to append self.COLL_UUID to filtered_path here.
137 filtered_path = Path(self.mounttmp, self.TAG_NAME)
139 filtered_path = Path(self.mounttmp, self.TAG_NAME, self.COLL_UUID, 'foo', 'nonexistent')
140 expect_path = filtered_path.parent
142 expect_path.exists(),
143 f"path not found but should exist: {expect_path}",
146 filtered_path.exists(),
147 f"path was found but should be filtered out: {filtered_path}",
151 class FiltersIntegrationTest(IntegrationTest):
152 COLLECTIONS_BY_PROP = {
153 coll['properties']['MainFile']: coll
154 for coll in _COLLECTIONS.values()
155 if coll['owner_uuid'] == _GROUPS['fuse_filters_test_project']['uuid']
157 PROP_VALUES = list(COLLECTIONS_BY_PROP)
159 for test_n, query in enumerate(['foo', 'ba?']):
160 @IntegrationTest.mount([
161 '--filters', json.dumps([
162 ['collections.properties.MainFile', 'like', query],
164 '--mount-by-pdh', 'by_pdh',
165 '--mount-by-id', 'by_id',
166 '--mount-home', 'home',
168 def _test_func(self, query=query):
169 pdh_path = Path(self.mnt, 'by_pdh')
170 id_path = Path(self.mnt, 'by_id')
171 home_path = Path(self.mnt, 'home')
172 query_re = re.compile(query.replace('?', '.'))
173 for prop_val, coll in self.COLLECTIONS_BY_PROP.items():
174 should_exist = query_re.fullmatch(prop_val) is not None
176 pdh_path / coll['portable_data_hash'],
177 id_path / coll['portable_data_hash'],
178 id_path / coll['uuid'],
179 home_path / coll['name'],
184 f"{path} from MainFile={prop_val} exists!={should_exist}",
186 exec(f"test_collection_properties_filters_{test_n} = _test_func")
188 for test_n, mount_opts in enumerate([
190 ['--project', _GROUPS['aproject']['uuid']],
192 @IntegrationTest.mount([
193 '--filters', json.dumps([
194 ['collections.name', 'like', 'zzzzz-4zz18-%'],
195 ['groups.name', 'like', 'A %roject'],
199 def _test_func(self, mount_opts=mount_opts):
200 root_path = Path(self.mnt)
201 root_depth = len(root_path.parts)
203 name_re = re.compile(r'(zzzzz-4zz18-.*|A .*roject)')
204 dir_queue = [root_path]
206 root_path = dir_queue.pop()
207 max_depth = max(max_depth, len(root_path.parts))
208 for child in root_path.iterdir():
209 if not child.is_dir():
211 match = name_re.fullmatch(child.name)
212 self.assertIsNotNone(
214 "found directory with name that should've been filtered",
216 if not match.group(1).startswith('zzzzz-4zz18-'):
217 dir_queue.append(child)
218 self.assertGreaterEqual(
220 root_depth + (2 if mount_opts[0] == '--home' else 1),
221 "test descended fewer subdirectories than expected",
223 exec(f"test_multiple_name_filters_{test_n} = _test_func")