Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/storage_tests.py
| Show All 17 Lines | |||||
| from swh.model import from_disk | from swh.model import from_disk | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.hypothesis_strategies import objects | from swh.model.hypothesis_strategies import objects | ||||
| from swh.model.identifiers import SWHID | from swh.model.identifiers import SWHID | ||||
| from swh.model.model import ( | from swh.model.model import ( | ||||
| Content, | Content, | ||||
| Directory, | Directory, | ||||
| ExtID, | |||||
| MetadataTargetType, | MetadataTargetType, | ||||
| ObjectType, | |||||
| Origin, | Origin, | ||||
| OriginVisit, | OriginVisit, | ||||
| OriginVisitStatus, | OriginVisitStatus, | ||||
| Person, | Person, | ||||
| Revision, | Revision, | ||||
| SkippedContent, | SkippedContent, | ||||
| Snapshot, | Snapshot, | ||||
| TargetType, | TargetType, | ||||
| ▲ Show 20 Lines • Show All 1,011 Lines • ▼ Show 20 Lines | def test_revision_get_random(self, swh_storage, sample_data): | ||||
| swh_storage.revision_add([revision1, revision2, revision3]) | swh_storage.revision_add([revision1, revision2, revision3]) | ||||
| assert swh_storage.revision_get_random() in { | assert swh_storage.revision_get_random() in { | ||||
| revision1.id, | revision1.id, | ||||
| revision2.id, | revision2.id, | ||||
| revision3.id, | revision3.id, | ||||
| } | } | ||||
| def test_extid_add_git(self, swh_storage, sample_data): | |||||
| gitids = [ | |||||
| revision.id | |||||
| for revision in sample_data.revisions | |||||
| if revision.type.value == "git" | |||||
| ] | |||||
| nullids = [None] * len(gitids) | |||||
| assert swh_storage.extid_get_from_extid("git", gitids) == nullids | |||||
| assert swh_storage.extid_get_from_swhid(ObjectType.REVISION, gitids) == nullids | |||||
| extids = [ | |||||
| ExtID( | |||||
| extid=gitid, | |||||
| extid_type="git", | |||||
| target=gitid, | |||||
| target_type=ObjectType.REVISION, | |||||
| ) | |||||
| for gitid in gitids | |||||
| ] | |||||
| summary = swh_storage.extid_add(extids) | |||||
| assert summary == {"extid:add": len(gitids)} | |||||
| assert swh_storage.extid_get_from_extid("git", gitids) == extids | |||||
| assert swh_storage.extid_get_from_swhid(ObjectType.REVISION, gitids) == extids | |||||
| assert swh_storage.extid_get_from_extid("hg", gitids) == nullids | |||||
| assert swh_storage.extid_get_from_swhid(ObjectType.RELEASE, gitids) == nullids | |||||
| def test_extid_add_hg(self, swh_storage, sample_data): | |||||
| def get_node(revision): | |||||
| node = None | |||||
| if revision.extra_headers: | |||||
| node = dict(revision.extra_headers).get(b"node") | |||||
| if node is None and revision.metadata: | |||||
| node = hash_to_bytes(revision.metadata.get("node")) | |||||
| return node | |||||
| swhids = [ | |||||
| revision.id | |||||
| for revision in sample_data.revisions | |||||
| if revision.type.value == "hg" | |||||
| ] | |||||
| extids = [ | |||||
| get_node(revision) | |||||
| for revision in sample_data.revisions | |||||
| 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_swhid(ObjectType.REVISION, swhids) == nullids | |||||
| extid_objs = [ | |||||
| ExtID( | |||||
| extid=hgid, | |||||
| extid_type="hg", | |||||
| target=swhid, | |||||
| target_type=ObjectType.REVISION, | |||||
| ) | |||||
| for hgid, swhid in zip(extids, swhids) | |||||
| ] | |||||
| summary = swh_storage.extid_add(extid_objs) | |||||
| assert summary == {"extid:add": len(swhids)} | |||||
| assert swh_storage.extid_get_from_extid("hg", extids) == extid_objs | |||||
| assert ( | |||||
| swh_storage.extid_get_from_swhid(ObjectType.REVISION, swhids) == extid_objs | |||||
| ) | |||||
| assert swh_storage.extid_get_from_extid("git", extids) == nullids | |||||
| assert swh_storage.extid_get_from_swhid(ObjectType.RELEASE, swhids) == nullids | |||||
| 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]) | ||||
| assert actual_result == {"release:add": 2} | assert actual_result == {"release:add": 2} | ||||
| ▲ Show 20 Lines • Show All 2,875 Lines • Show Last 20 Lines | |||||