Changeset View
Changeset View
Standalone View
Standalone View
swh/fuse/tests/test_revision.py
| import json | import json | ||||
| import os | import os | ||||
| import time | |||||
| from swh.fuse.fs.artifact import RevisionHistoryShardByPage | 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, | ||||
| ) | ) | ||||
| Show All 25 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" | dir_path = fuse_mntdir / "archive" / REV_SMALL_HISTORY / "history" | ||||
| assert os.listdir(dir_path) == ["by-hash", "by-page"] | assert os.listdir(dir_path) == ["by-date", "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( | expected = set( | ||||
| Show All 9 Lines | def test_list_history(fuse_mntdir): | ||||
| dir_by_page = dir_path / "by-page" | dir_by_page = dir_path / "by-page" | ||||
| for idx, swhid in enumerate(expected): | for idx, swhid in enumerate(expected): | ||||
| page_number = idx // RevisionHistoryShardByPage.PAGE_SIZE | page_number = idx // RevisionHistoryShardByPage.PAGE_SIZE | ||||
| depth1 = RevisionHistoryShardByPage.PAGE_FMT.format(page_number=page_number) | depth1 = RevisionHistoryShardByPage.PAGE_FMT.format(page_number=page_number) | ||||
| depth2 = str(swhid) | depth2 = str(swhid) | ||||
| assert (dir_by_page / depth1).exists() | assert (dir_by_page / depth1).exists() | ||||
| assert depth2 in (os.listdir(dir_by_page / depth1)) | assert depth2 in (os.listdir(dir_by_page / depth1)) | ||||
| dir_by_date = dir_path / "by-date" | |||||
| # Wait max 1 second to populate by-date/ dir | |||||
| for i in range(100): | |||||
| if ".status" not in os.listdir(dir_by_date): | |||||
| break | |||||
| time.sleep(0.01) | |||||
| assert os.listdir(dir_by_date) == ["2010"] | |||||
| assert os.listdir(dir_by_date / "2010") == ["06"] | |||||
| assert os.listdir(dir_by_date / "2010/06") == ["25", "24", "23", "16"] | |||||