Changeset View
Changeset View
Standalone View
Standalone View
swh/vault/tests/test_to_disk.py
| Show First 20 Lines • Show All 64 Lines • ▼ Show 20 Lines | def test_get_filtered_files_content__unknown_status(swh_storage): | ||||
| files_data = [ | files_data = [ | ||||
| { | { | ||||
| "status": "visible", | "status": "visible", | ||||
| "sha1": content.sha1, | "sha1": content.sha1, | ||||
| "sha1_git": content.sha1_git, | "sha1_git": content.sha1_git, | ||||
| "target": content.sha1_git, | "target": content.sha1_git, | ||||
| }, | }, | ||||
| { | { | ||||
| "status": None, | "status": "blah", | ||||
| "target": b"c" * 20, | "target": b"c" * 20, | ||||
| }, | }, | ||||
| ] | ] | ||||
| with pytest.raises(AssertionError, match="unexpected status None"): | with pytest.raises(AssertionError, match="unexpected status 'blah'"): | ||||
| list(get_filtered_files_content(swh_storage, files_data)) | list(get_filtered_files_content(swh_storage, files_data)) | ||||
| def test_directory_builder(swh_storage, tmp_path): | def _fill_storage(swh_storage, exclude_cnt3=False): | ||||
| cnt1 = Content.from_data(b"foo bar") | cnt1 = Content.from_data(b"foo bar") | ||||
| cnt2 = Content.from_data(b"bar baz") | cnt2 = Content.from_data(b"bar baz") | ||||
| cnt3 = Content.from_data(b"baz qux") | cnt3 = Content.from_data(b"baz qux") | ||||
| dir1 = Directory( | dir1 = Directory( | ||||
| entries=( | entries=( | ||||
| DirectoryEntry( | DirectoryEntry( | ||||
| name=b"content1", | name=b"content1", | ||||
| type="file", | type="file", | ||||
| Show All 19 Lines | dir2 = Directory( | ||||
| DirectoryEntry( | DirectoryEntry( | ||||
| name=b"subdirectory", | name=b"subdirectory", | ||||
| type="dir", | type="dir", | ||||
| target=dir1.id, | target=dir1.id, | ||||
| perms=DentryPerms.directory, | perms=DentryPerms.directory, | ||||
| ), | ), | ||||
| ) | ) | ||||
| ) | ) | ||||
| if exclude_cnt3: | |||||
| swh_storage.content_add([cnt1, cnt2]) | |||||
| else: | |||||
| swh_storage.content_add([cnt1, cnt2, cnt3]) | swh_storage.content_add([cnt1, cnt2, cnt3]) | ||||
| swh_storage.directory_add([dir1, dir2]) | swh_storage.directory_add([dir1, dir2]) | ||||
| return dir2 | |||||
| def test_directory_builder(swh_storage, tmp_path): | |||||
| dir2 = _fill_storage(swh_storage) | |||||
| root = tmp_path / "root" | root = tmp_path / "root" | ||||
| builder = DirectoryBuilder(swh_storage, bytes(root), dir2.id) | builder = DirectoryBuilder(swh_storage, bytes(root), dir2.id) | ||||
| assert not root.exists() | assert not root.exists() | ||||
| builder.build() | builder.build() | ||||
| assert root.is_dir() | assert root.is_dir() | ||||
| assert set(root.glob("**/*")) == { | assert set(root.glob("**/*")) == { | ||||
| root / "subdirectory", | root / "subdirectory", | ||||
| root / "subdirectory" / "content1", | root / "subdirectory" / "content1", | ||||
| root / "subdirectory" / "content2", | root / "subdirectory" / "content2", | ||||
| root / "content3", | root / "content3", | ||||
| } | } | ||||
| assert (root / "subdirectory" / "content1").open().read() == "foo bar" | |||||
| assert (root / "subdirectory" / "content2").open().read() == "bar baz" | |||||
| assert (root / "content3").open().read() == "baz qux" | |||||
| def test_directory_builder_missing_content(swh_storage, tmp_path): | |||||
| dir2 = _fill_storage(swh_storage, exclude_cnt3=True) | |||||
| root = tmp_path / "root" | |||||
| builder = DirectoryBuilder(swh_storage, bytes(root), dir2.id) | |||||
| assert not root.exists() | |||||
| builder.build() | |||||
| assert root.is_dir() | |||||
| assert "This content is missing" in (root / "content3").open().read() | |||||