Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/tests/test_cookers.py
| Show All 11 Lines | |||||
| import pathlib | import pathlib | ||||
| import shutil | import shutil | ||||
| import subprocess | import subprocess | ||||
| import tarfile | import tarfile | ||||
| import tempfile | import tempfile | ||||
| import unittest | import unittest | ||||
| import unittest.mock | import unittest.mock | ||||
| import attrs | import attr | ||||
| import dulwich.fastexport | import dulwich.fastexport | ||||
| import dulwich.index | import dulwich.index | ||||
| import dulwich.objects | import dulwich.objects | ||||
| import dulwich.porcelain | import dulwich.porcelain | ||||
| import dulwich.repo | import dulwich.repo | ||||
| import pytest | import pytest | ||||
| from swh.loader.git.from_disk import GitLoaderFromDisk | from swh.loader.git.from_disk import GitLoaderFromDisk | ||||
| ▲ Show 20 Lines • Show All 408 Lines • ▼ Show 20 Lines | def test_directory_filtered_objects(self, git_loader, cook_extract_directory): | ||||
| obj_id_hex = repo.repo[c].tree.decode() | obj_id_hex = repo.repo[c].tree.decode() | ||||
| obj_id = hashutil.hash_to_bytes(obj_id_hex) | obj_id = hashutil.hash_to_bytes(obj_id_hex) | ||||
| swhid = CoreSWHID(object_type=ObjectType.DIRECTORY, object_id=obj_id) | swhid = CoreSWHID(object_type=ObjectType.DIRECTORY, object_id=obj_id) | ||||
| # alter the content of the storage | # alter the content of the storage | ||||
| # 1/ make file 2 an hidden file object | # 1/ make file 2 an hidden file object | ||||
| loader.storage._allow_overwrite = True | loader.storage._allow_overwrite = True | ||||
| cnt2 = attrs.evolve( | cnt2 = attr.evolve( | ||||
| loader.storage.content_get([id_2])[0], status="hidden", data=file_2 | loader.storage.content_get([id_2])[0], status="hidden", data=file_2 | ||||
| ) | ) | ||||
| loader.storage.content_add([cnt2]) | loader.storage.content_add([cnt2]) | ||||
| assert loader.storage.content_get([id_2])[0].status == "hidden" | assert loader.storage.content_get([id_2])[0].status == "hidden" | ||||
| # 2/ make file 3 an skipped file object | # 2/ make file 3 an skipped file object | ||||
| cnt3 = loader.storage.content_get([id_3])[0].to_dict() | cnt3 = loader.storage.content_get([id_3])[0].to_dict() | ||||
| cnt3["status"] = "absent" | cnt3["status"] = "absent" | ||||
| ▲ Show 20 Lines • Show All 367 Lines • ▼ Show 20 Lines | def load_repo_filtered_objects(self, git_loader): | ||||
| obj_id = hashutil.hash_to_bytes(obj_id_hex) | obj_id = hashutil.hash_to_bytes(obj_id_hex) | ||||
| swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=obj_id) | swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=obj_id) | ||||
| loader = git_loader(str(rp)) | loader = git_loader(str(rp)) | ||||
| loader.load() | loader.load() | ||||
| # alter the content of the storage | # alter the content of the storage | ||||
| # 1/ make file 2 an hidden file object | # 1/ make file 2 an hidden file object | ||||
| loader.storage._allow_overwrite = True | loader.storage._allow_overwrite = True | ||||
| cnt2 = attrs.evolve( | cnt2 = attr.evolve( | ||||
| loader.storage.content_get([id_2])[0], status="hidden", data=file_2 | loader.storage.content_get([id_2])[0], status="hidden", data=file_2 | ||||
| ) | ) | ||||
| loader.storage.content_add([cnt2]) | loader.storage.content_add([cnt2]) | ||||
| assert loader.storage.content_get([id_2])[0].status == "hidden" | assert loader.storage.content_get([id_2])[0].status == "hidden" | ||||
| # 2/ make file 3 an skipped file object | # 2/ make file 3 an skipped file object | ||||
| cnt3 = loader.storage.content_get([id_3])[0].to_dict() | cnt3 = loader.storage.content_get([id_3])[0].to_dict() | ||||
| cnt3["status"] = "absent" | cnt3["status"] = "absent" | ||||
| ▲ Show 20 Lines • Show All 363 Lines • Show Last 20 Lines | |||||