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, | ||||
} | } |