Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
| Show First 20 Lines • Show All 1,084 Lines • ▼ Show 20 Lines | class TestStorage: | ||||
| def test_extid_add_git(self, swh_storage, sample_data): | def test_extid_add_git(self, swh_storage, sample_data): | ||||
| gitids = [ | gitids = [ | ||||
| revision.id | revision.id | ||||
| for revision in sample_data.revisions | for revision in sample_data.revisions | ||||
| if revision.type.value == "git" | if revision.type.value == "git" | ||||
| ] | ] | ||||
| nullids = [None] * len(gitids) | |||||
| extids = [ | extids = [ | ||||
| ExtID( | ExtID( | ||||
| extid=gitid, | extid=gitid, | ||||
| extid_type="git", | extid_type="git", | ||||
| target=CoreSWHID(object_id=gitid, object_type=ObjectType.REVISION,), | target=CoreSWHID(object_id=gitid, object_type=ObjectType.REVISION,), | ||||
| ) | ) | ||||
| for gitid in gitids | for gitid in gitids | ||||
| ] | ] | ||||
| assert swh_storage.extid_get_from_extid("git", gitids) == nullids | assert swh_storage.extid_get_from_extid("git", gitids) == [] | ||||
| assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == nullids | assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == [] | ||||
| summary = swh_storage.extid_add(extids) | summary = swh_storage.extid_add(extids) | ||||
| assert summary == {"extid:add": len(gitids)} | assert summary == {"extid:add": len(gitids)} | ||||
| assert swh_storage.extid_get_from_extid("git", gitids) == extids | assert swh_storage.extid_get_from_extid("git", gitids) == extids | ||||
| assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == extids | assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == extids | ||||
| assert swh_storage.extid_get_from_extid("hg", gitids) == nullids | assert swh_storage.extid_get_from_extid("hg", gitids) == [] | ||||
| assert swh_storage.extid_get_from_target(ObjectType.RELEASE, gitids) == nullids | assert swh_storage.extid_get_from_target(ObjectType.RELEASE, gitids) == [] | ||||
| def test_extid_add_hg(self, swh_storage, sample_data): | def test_extid_add_hg(self, swh_storage, sample_data): | ||||
| def get_node(revision): | def get_node(revision): | ||||
| node = None | node = None | ||||
| if revision.extra_headers: | if revision.extra_headers: | ||||
| node = dict(revision.extra_headers).get(b"node") | node = dict(revision.extra_headers).get(b"node") | ||||
| if node is None and revision.metadata: | if node is None and revision.metadata: | ||||
| node = hash_to_bytes(revision.metadata.get("node")) | node = hash_to_bytes(revision.metadata.get("node")) | ||||
| return node | return node | ||||
| swhids = [ | swhids = [ | ||||
| revision.id | revision.id | ||||
| for revision in sample_data.revisions | for revision in sample_data.revisions | ||||
| if revision.type.value == "hg" | if revision.type.value == "hg" | ||||
| ] | ] | ||||
| extids = [ | extids = [ | ||||
| get_node(revision) | get_node(revision) | ||||
| for revision in sample_data.revisions | for revision in sample_data.revisions | ||||
| if revision.type.value == "hg" | if revision.type.value == "hg" | ||||
| ] | ] | ||||
| nullids = [None] * len(swhids) | |||||
| assert swh_storage.extid_get_from_extid("hg", extids) == nullids | assert swh_storage.extid_get_from_extid("hg", extids) == [] | ||||
| assert swh_storage.extid_get_from_target(ObjectType.REVISION, swhids) == nullids | assert swh_storage.extid_get_from_target(ObjectType.REVISION, swhids) == [] | ||||
| extid_objs = [ | extid_objs = [ | ||||
| ExtID( | ExtID( | ||||
| extid=hgid, | extid=hgid, | ||||
| extid_type="hg", | extid_type="hg", | ||||
| target=CoreSWHID(object_id=swhid, object_type=ObjectType.REVISION,), | target=CoreSWHID(object_id=swhid, object_type=ObjectType.REVISION,), | ||||
| ) | ) | ||||
| for hgid, swhid in zip(extids, swhids) | for hgid, swhid in zip(extids, swhids) | ||||
| ] | ] | ||||
| summary = swh_storage.extid_add(extid_objs) | summary = swh_storage.extid_add(extid_objs) | ||||
| assert summary == {"extid:add": len(swhids)} | assert summary == {"extid:add": len(swhids)} | ||||
| assert swh_storage.extid_get_from_extid("hg", extids) == extid_objs | assert swh_storage.extid_get_from_extid("hg", extids) == extid_objs | ||||
| assert ( | assert ( | ||||
| swh_storage.extid_get_from_target(ObjectType.REVISION, swhids) == extid_objs | swh_storage.extid_get_from_target(ObjectType.REVISION, swhids) == extid_objs | ||||
| ) | ) | ||||
| assert swh_storage.extid_get_from_extid("git", extids) == nullids | assert swh_storage.extid_get_from_extid("git", extids) == [] | ||||
| assert swh_storage.extid_get_from_target(ObjectType.RELEASE, swhids) == nullids | assert swh_storage.extid_get_from_target(ObjectType.RELEASE, swhids) == [] | ||||
| def test_extid_add_twice(self, swh_storage, sample_data): | def test_extid_add_twice(self, swh_storage, sample_data): | ||||
| gitids = [ | gitids = [ | ||||
| revision.id | revision.id | ||||
| for revision in sample_data.revisions | for revision in sample_data.revisions | ||||
| if revision.type.value == "git" | if revision.type.value == "git" | ||||
| ] | ] | ||||
| Show All 10 Lines | def test_extid_add_twice(self, swh_storage, sample_data): | ||||
| assert summary == {"extid:add": len(gitids)} | assert summary == {"extid:add": len(gitids)} | ||||
| # add them again, should be noop | # add them again, should be noop | ||||
| summary = swh_storage.extid_add(extids) | summary = swh_storage.extid_add(extids) | ||||
| # assert summary == {"extid:add": 0} | # assert summary == {"extid:add": 0} | ||||
| assert swh_storage.extid_get_from_extid("git", gitids) == extids | assert swh_storage.extid_get_from_extid("git", gitids) == extids | ||||
| assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == extids | assert swh_storage.extid_get_from_target(ObjectType.REVISION, gitids) == extids | ||||
| def test_extid_add_extid_unicity(self, swh_storage, sample_data): | def test_extid_add_extid_multicity(self, swh_storage, sample_data): | ||||
| ids = [ | ids = [ | ||||
| revision.id | revision.id | ||||
| for revision in sample_data.revisions | for revision in sample_data.revisions | ||||
| if revision.type.value == "git" | if revision.type.value == "git" | ||||
| ] | ] | ||||
| nullids = [None] * len(ids) | |||||
| extids = [ | extids = [ | ||||
| ExtID( | ExtID( | ||||
| extid=extid, | extid=extid, | ||||
| extid_type="git", | extid_type="git", | ||||
| target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | ||||
| ) | ) | ||||
| for extid in ids | for extid in ids | ||||
| ] | ] | ||||
| swh_storage.extid_add(extids) | swh_storage.extid_add(extids) | ||||
| # try to add "modified-extid" versions, should be noops | # try to add "modified-extid" versions, should be added | ||||
| extids2 = [ | extids2 = [ | ||||
| ExtID( | ExtID( | ||||
| extid=extid, | extid=extid, | ||||
| extid_type="hg", | extid_type="hg", | ||||
| target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | ||||
| ) | ) | ||||
| for extid in ids | for extid in ids | ||||
| ] | ] | ||||
| swh_storage.extid_add(extids2) | swh_storage.extid_add(extids2) | ||||
| assert swh_storage.extid_get_from_extid("git", ids) == extids | assert swh_storage.extid_get_from_extid("git", ids) == extids | ||||
| assert swh_storage.extid_get_from_extid("hg", ids) == nullids | assert swh_storage.extid_get_from_extid("hg", ids) == extids2 | ||||
| assert swh_storage.extid_get_from_target(ObjectType.REVISION, ids) == extids | assert set(swh_storage.extid_get_from_target(ObjectType.REVISION, ids)) == { | ||||
| *extids, | |||||
| *extids2, | |||||
| } | |||||
| def test_extid_add_target_unicity(self, swh_storage, sample_data): | def test_extid_add_target_multicity(self, swh_storage, sample_data): | ||||
| ids = [ | ids = [ | ||||
| revision.id | revision.id | ||||
| for revision in sample_data.revisions | for revision in sample_data.revisions | ||||
| if revision.type.value == "git" | if revision.type.value == "git" | ||||
| ] | ] | ||||
| nullids = [None] * len(ids) | |||||
| extids = [ | extids = [ | ||||
| ExtID( | ExtID( | ||||
| extid=extid, | extid=extid, | ||||
| extid_type="git", | extid_type="git", | ||||
| target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | target=CoreSWHID(object_id=extid, object_type=ObjectType.REVISION,), | ||||
| ) | ) | ||||
| for extid in ids | for extid in ids | ||||
| ] | ] | ||||
| swh_storage.extid_add(extids) | swh_storage.extid_add(extids) | ||||
| # try to add "modified" versions, should be noops | # try to add "modified" versions, should be added | ||||
| extids2 = [ | extids2 = [ | ||||
| ExtID( | ExtID( | ||||
| extid=extid, | extid=extid, | ||||
| extid_type="git", | extid_type="git", | ||||
| target=CoreSWHID(object_id=extid, object_type=ObjectType.RELEASE,), | target=CoreSWHID(object_id=extid, object_type=ObjectType.RELEASE,), | ||||
| ) | ) | ||||
| for extid in ids | for extid in ids | ||||
| ] | ] | ||||
| swh_storage.extid_add(extids2) | swh_storage.extid_add(extids2) | ||||
| assert swh_storage.extid_get_from_extid("git", ids) == extids | assert set(swh_storage.extid_get_from_extid("git", ids)) == {*extids, *extids2} | ||||
| assert swh_storage.extid_get_from_target(ObjectType.REVISION, ids) == extids | assert swh_storage.extid_get_from_target(ObjectType.REVISION, ids) == extids | ||||
| assert swh_storage.extid_get_from_target(ObjectType.RELEASE, ids) == nullids | assert swh_storage.extid_get_from_target(ObjectType.RELEASE, ids) == extids2 | ||||
| def test_release_add(self, swh_storage, sample_data): | def test_release_add(self, swh_storage, sample_data): | ||||
| release, release2 = sample_data.releases[:2] | release, release2 = sample_data.releases[:2] | ||||
| init_missing = swh_storage.release_missing([release.id, release2.id]) | init_missing = swh_storage.release_missing([release.id, release2.id]) | ||||
| assert list(init_missing) == [release.id, release2.id] | assert list(init_missing) == [release.id, release2.id] | ||||
| actual_result = swh_storage.release_add([release, release2]) | actual_result = swh_storage.release_add([release, release2]) | ||||
| ▲ Show 20 Lines • Show All 3,091 Lines • Show Last 20 Lines | |||||