Changeset View
Changeset View
Standalone View
Standalone View
swh/fuse/tests/test_revision.py
| import json | import json | ||||
| import os | import os | ||||
| from swh.fuse.fs.artifact import RevisionHistoryShardByPage | |||||
| from swh.fuse.tests.api_url import GRAPH_API_REQUEST | from swh.fuse.tests.api_url import GRAPH_API_REQUEST | ||||
| from swh.fuse.tests.common import ( | from swh.fuse.tests.common import ( | ||||
| check_dir_name_entries, | check_dir_name_entries, | ||||
| get_data_from_graph_archive, | get_data_from_graph_archive, | ||||
| get_data_from_web_archive, | get_data_from_web_archive, | ||||
| ) | ) | ||||
| from swh.fuse.tests.data.config import REV_SMALL_HISTORY, ROOT_DIR, ROOT_REV | from swh.fuse.tests.data.config import REV_SMALL_HISTORY, ROOT_DIR, ROOT_REV | ||||
| from swh.model.identifiers import parse_swhid | from swh.model.identifiers import parse_swhid | ||||
| Show All 22 Lines | |||||
| def test_list_parent(fuse_mntdir): | def test_list_parent(fuse_mntdir): | ||||
| file_path = fuse_mntdir / "archive" / ROOT_REV / "parent" | file_path = fuse_mntdir / "archive" / ROOT_REV / "parent" | ||||
| assert file_path.is_symlink() | assert file_path.is_symlink() | ||||
| assert os.readlink(file_path) == "parents/1/" | assert os.readlink(file_path) == "parents/1/" | ||||
| def test_list_history(fuse_mntdir): | def test_list_history(fuse_mntdir): | ||||
| dir_path = fuse_mntdir / "archive" / REV_SMALL_HISTORY / "history/by-hash" | dir_path = fuse_mntdir / "archive" / REV_SMALL_HISTORY / "history" | ||||
| assert os.listdir(dir_path) == ["by-hash", "by-page"] | |||||
| history_meta = get_data_from_graph_archive( | history_meta = get_data_from_graph_archive( | ||||
| REV_SMALL_HISTORY, GRAPH_API_REQUEST.HISTORY | REV_SMALL_HISTORY, GRAPH_API_REQUEST.HISTORY | ||||
| ) | ) | ||||
| history = history_meta.strip() | history = history_meta.strip() | ||||
| # Only keep second node in the edge because first node is redundant | # Only keep second node in the edge because first node is redundant | ||||
| # information or the root node (hence not an ancestor) | # information or the root node (hence not an ancestor) | ||||
| expected = set([edge.split(" ")[1] for edge in history.split("\n")]) | expected = set( | ||||
| map(parse_swhid, [edge.split(" ")[1] for edge in history.split("\n")]) | |||||
| ) | |||||
| dir_by_hash = dir_path / "by-hash" | |||||
| for swhid in expected: | for swhid in expected: | ||||
| swhid = parse_swhid(swhid) | |||||
| depth1 = swhid.object_id[:2] | depth1 = swhid.object_id[:2] | ||||
| depth2 = str(swhid) | depth2 = str(swhid) | ||||
| assert (dir_path / depth1).exists() | assert (dir_by_hash / depth1).exists() | ||||
| assert depth2 in (os.listdir(dir_path / depth1)) | assert depth2 in (os.listdir(dir_by_hash / depth1)) | ||||
| dir_by_page = dir_path / "by-page" | |||||
| for idx, swhid in enumerate(expected): | |||||
| page_number = idx // RevisionHistoryShardByPage.PAGE_SIZE | |||||
| depth1 = RevisionHistoryShardByPage.PAGE_FMT.format(page_number=page_number) | |||||
| depth2 = str(swhid) | |||||
| assert (dir_by_page / depth1).exists() | |||||
| assert depth2 in (os.listdir(dir_by_page / depth1)) | |||||