Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_multiplexer_filter.py
| Show First 20 Lines • Show All 102 Lines • ▼ Show 20 Lines | def setUp(self): | ||||
| super().setUp() | super().setUp() | ||||
| # Use a hack here : as the mock uses the content as id, it is easy to | # Use a hack here : as the mock uses the content as id, it is easy to | ||||
| # create contents that are filtered or not. | # create contents that are filtered or not. | ||||
| self.prefix = "71" | self.prefix = "71" | ||||
| self.tmpdir = tempfile.mkdtemp() | self.tmpdir = tempfile.mkdtemp() | ||||
| # Make the storage filtered | # Make the storage filtered | ||||
| self.sconf = { | self.sconf = { | ||||
| "cls": "pathslicing", | "cls": "pathslicing", | ||||
| "args": {"root": self.tmpdir, "slicing": "0:5"}, | "root": self.tmpdir, | ||||
| "slicing": "0:5", | |||||
| } | } | ||||
| storage = get_objstorage(**self.sconf) | storage = get_objstorage(**self.sconf) | ||||
| self.base_storage = storage | self.base_storage = storage | ||||
| self.storage = self.filter_storage(self.sconf) | self.storage = self.filter_storage(self.sconf) | ||||
| # Present content with valid id | # Present content with valid id | ||||
| self.present_valid_content = self.ensure_valid(b"yroqdtotji") | self.present_valid_content = self.ensure_valid(b"yroqdtotji") | ||||
| self.present_valid_id = compute_hash(self.present_valid_content) | self.present_valid_id = compute_hash(self.present_valid_content) | ||||
| ▲ Show 20 Lines • Show All 190 Lines • ▼ Show 20 Lines | def ensure_invalid(self, content): | ||||
| obj_id = compute_hash(content) | obj_id = compute_hash(content) | ||||
| hex_obj_id = hashutil.hash_to_hex(obj_id) | hex_obj_id = hashutil.hash_to_hex(obj_id) | ||||
| self.assertFalse(hex_obj_id.startswith(self.prefix)) | self.assertFalse(hex_obj_id.startswith(self.prefix)) | ||||
| return content | return content | ||||
| def filter_storage(self, sconf): | def filter_storage(self, sconf): | ||||
| return get_objstorage( | return get_objstorage( | ||||
| "filtered", | "filtered", | ||||
| {"storage_conf": sconf, "filters_conf": [id_prefix(self.prefix)]}, | storage_conf=sconf, | ||||
| filters_conf=[id_prefix(self.prefix)], | |||||
| ) | ) | ||||
| class TestRegexFilter(MixinTestIdFilter, unittest.TestCase): | class TestRegexFilter(MixinTestIdFilter, unittest.TestCase): | ||||
| def setUp(self): | def setUp(self): | ||||
| self.regex = r"[a-f][0-9].*" | self.regex = r"[a-f][0-9].*" | ||||
| super().setUp() | super().setUp() | ||||
| def filter_storage(self, sconf): | def filter_storage(self, sconf): | ||||
| return get_objstorage( | return get_objstorage( | ||||
| "filtered", {"storage_conf": sconf, "filters_conf": [id_regex(self.regex)]} | "filtered", storage_conf=sconf, filters_conf=[id_regex(self.regex)] | ||||
| ) | ) | ||||