Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_filter.py
| Show All 15 Lines | storage_config = { | ||||
| "cls": "pipeline", | "cls": "pipeline", | ||||
| "steps": [{"cls": "filter"}, {"cls": "memory"},], | "steps": [{"cls": "filter"}, {"cls": "memory"},], | ||||
| } | } | ||||
| return get_storage(**storage_config) | return get_storage(**storage_config) | ||||
| def test_filtering_proxy_storage_content(swh_storage, sample_data): | def test_filtering_proxy_storage_content(swh_storage, sample_data): | ||||
| sample_content = sample_data["content"][0] | sample_content = sample_data.content | ||||
| content = next(swh_storage.content_get([sample_content.sha1])) | content = next(swh_storage.content_get([sample_content.sha1])) | ||||
| assert not content | assert not content | ||||
| s = swh_storage.content_add([sample_content]) | s = swh_storage.content_add([sample_content]) | ||||
| assert s == { | assert s == { | ||||
| "content:add": 1, | "content:add": 1, | ||||
| "content:add:bytes": sample_content.length, | "content:add:bytes": sample_content.length, | ||||
| } | } | ||||
| content = next(swh_storage.content_get([sample_content.sha1])) | content = next(swh_storage.content_get([sample_content.sha1])) | ||||
| assert content is not None | assert content is not None | ||||
| s = swh_storage.content_add([sample_content]) | s = swh_storage.content_add([sample_content]) | ||||
| assert s == { | assert s == { | ||||
| "content:add": 0, | "content:add": 0, | ||||
| "content:add:bytes": 0, | "content:add:bytes": 0, | ||||
| } | } | ||||
| def test_filtering_proxy_storage_skipped_content(swh_storage, sample_data): | def test_filtering_proxy_storage_skipped_content(swh_storage, sample_data): | ||||
| sample_content = sample_data["skipped_content"][0] | sample_content = sample_data.skipped_content | ||||
| sample_content_dict = sample_content.to_dict() | sample_content_dict = sample_content.to_dict() | ||||
| content = next(swh_storage.skipped_content_missing([sample_content_dict])) | content = next(swh_storage.skipped_content_missing([sample_content_dict])) | ||||
| assert content["sha1"] == sample_content.sha1 | assert content["sha1"] == sample_content.sha1 | ||||
| s = swh_storage.skipped_content_add([sample_content]) | s = swh_storage.skipped_content_add([sample_content]) | ||||
| assert s == { | assert s == { | ||||
| "skipped_content:add": 1, | "skipped_content:add": 1, | ||||
| } | } | ||||
| content = list(swh_storage.skipped_content_missing([sample_content_dict])) | content = list(swh_storage.skipped_content_missing([sample_content_dict])) | ||||
| assert content == [] | assert content == [] | ||||
| s = swh_storage.skipped_content_add([sample_content]) | s = swh_storage.skipped_content_add([sample_content]) | ||||
| assert s == { | assert s == { | ||||
| "skipped_content:add": 0, | "skipped_content:add": 0, | ||||
| } | } | ||||
| def test_filtering_proxy_storage_skipped_content_missing_sha1_git( | def test_filtering_proxy_storage_skipped_content_missing_sha1_git( | ||||
| swh_storage, sample_data | swh_storage, sample_data | ||||
| ): | ): | ||||
| sample_contents = [ | sample_contents = [ | ||||
| attr.evolve(c, sha1_git=None) for c in sample_data["skipped_content"] | attr.evolve(c, sha1_git=None) for c in sample_data.skipped_contents | ||||
| ] | ] | ||||
| sample_content, sample_content2 = [c.to_dict() for c in sample_contents[:2]] | sample_content, sample_content2 = [c.to_dict() for c in sample_contents[:2]] | ||||
| content = next(swh_storage.skipped_content_missing([sample_content])) | content = next(swh_storage.skipped_content_missing([sample_content])) | ||||
| assert content["sha1"] == sample_content["sha1"] | assert content["sha1"] == sample_content["sha1"] | ||||
| s = swh_storage.skipped_content_add([sample_contents[0]]) | s = swh_storage.skipped_content_add([sample_contents[0]]) | ||||
| assert s == { | assert s == { | ||||
| "skipped_content:add": 1, | "skipped_content:add": 1, | ||||
| } | } | ||||
| content = list(swh_storage.skipped_content_missing([sample_content])) | content = list(swh_storage.skipped_content_missing([sample_content])) | ||||
| assert content == [] | assert content == [] | ||||
| s = swh_storage.skipped_content_add([sample_contents[1]]) | s = swh_storage.skipped_content_add([sample_contents[1]]) | ||||
| assert s == { | assert s == { | ||||
| "skipped_content:add": 1, | "skipped_content:add": 1, | ||||
| } | } | ||||
| content = list(swh_storage.skipped_content_missing([sample_content2])) | content = list(swh_storage.skipped_content_missing([sample_content2])) | ||||
| assert content == [] | assert content == [] | ||||
| def test_filtering_proxy_storage_revision(swh_storage, sample_data): | def test_filtering_proxy_storage_revision(swh_storage, sample_data): | ||||
| sample_revision = sample_data["revision"][0] | sample_revision = sample_data.revision | ||||
| revision = next(swh_storage.revision_get([sample_revision.id])) | revision = next(swh_storage.revision_get([sample_revision.id])) | ||||
| assert not revision | assert not revision | ||||
| s = swh_storage.revision_add([sample_revision]) | s = swh_storage.revision_add([sample_revision]) | ||||
| assert s == { | assert s == { | ||||
| "revision:add": 1, | "revision:add": 1, | ||||
| } | } | ||||
| revision = next(swh_storage.revision_get([sample_revision.id])) | revision = next(swh_storage.revision_get([sample_revision.id])) | ||||
| assert revision is not None | assert revision is not None | ||||
| s = swh_storage.revision_add([sample_revision]) | s = swh_storage.revision_add([sample_revision]) | ||||
| assert s == { | assert s == { | ||||
| "revision:add": 0, | "revision:add": 0, | ||||
| } | } | ||||
| def test_filtering_proxy_storage_directory(swh_storage, sample_data): | def test_filtering_proxy_storage_directory(swh_storage, sample_data): | ||||
| sample_directory = sample_data["directory"][0] | sample_directory = sample_data.directory | ||||
| directory = next(swh_storage.directory_missing([sample_directory.id])) | directory = next(swh_storage.directory_missing([sample_directory.id])) | ||||
| assert directory | assert directory | ||||
| s = swh_storage.directory_add([sample_directory]) | s = swh_storage.directory_add([sample_directory]) | ||||
| assert s == { | assert s == { | ||||
| "directory:add": 1, | "directory:add": 1, | ||||
| } | } | ||||
| directory = list(swh_storage.directory_missing([sample_directory.id])) | directory = list(swh_storage.directory_missing([sample_directory.id])) | ||||
| assert not directory | assert not directory | ||||
| s = swh_storage.directory_add([sample_directory]) | s = swh_storage.directory_add([sample_directory]) | ||||
| assert s == { | assert s == { | ||||
| "directory:add": 0, | "directory:add": 0, | ||||
| } | } | ||||