def test_load_extids() -> None: """Checks PackageLoader.load() skips iff it should, and writes (only) the new ExtIDs""" storage = get_storage("memory") origin = "http://example.org" rev1_swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=b"a" * 20) rev2_swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=b"b" * 20) rev3_swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=b"c" * 20) rev4_swhid = CoreSWHID(object_type=ObjectType.REVISION, object_id=b"d" * 20) dir_swhid = CoreSWHID(object_type=ObjectType.DIRECTORY, object_id=b"e" * 20) loader = StubPackageLoader(storage, "http://example.org") patch.object( loader, "_load_revision", return_value=(rev4_swhid.object_id, dir_swhid.object_id), autospec=True, ).start() # Results of a previous load storage.extid_add( [ ExtID("extid-type1", b"extid-of-v1.0", rev1_swhid), ExtID("extid-type2", b"extid-of-v2.0", rev2_swhid), ] ) last_snapshot = Snapshot( branches={ b"v1.0": SnapshotBranch( target_type=TargetType.REVISION, target=rev1_swhid.object_id ), b"v2.0": SnapshotBranch( target_type=TargetType.REVISION, target=rev2_swhid.object_id ), b"v3.0": SnapshotBranch( target_type=TargetType.REVISION, target=rev3_swhid.object_id ), } ) storage.snapshot_add([last_snapshot]) date = datetime.datetime.now(tz=datetime.timezone.utc) storage.origin_add([Origin(url=origin)]) storage.origin_visit_add( [OriginVisit(origin="http://example.org", visit=1, date=date, type="tar")] ) storage.origin_visit_status_add( [ OriginVisitStatus( origin=origin, visit=1, status="full", date=date, snapshot=last_snapshot.id, ) ] ) loader.load() assert loader._load_revision.mock_calls == [ # type: ignore # v1.0: not loaded because there is already its (extid_type, extid, rev) # in the storage. # v2.0: loaded, because there is already a similar extid, but different type call(StubPackageInfo(origin, "example-v2.0.tar"), Origin(url=origin)), # v3.0: loaded despite having an (extid_type, extid) in storage, because # the target of the extid is not in the previous snapshot call(StubPackageInfo(origin, "example-v3.0.tar"), Origin(url=origin)), # v4.0: loaded, because there isn't its extid call(StubPackageInfo(origin, "example-v4.0.tar"), Origin(url=origin)), ] # then check the snapshot has all the branches. # versions 2.0 to 4.0 all point to rev4_swhid (instead of the value of the last # snapshot), because they had to be loaded (mismatched extid), and the mocked # _load_revision always returns rev4_swhid. snapshot = Snapshot( branches={ b"branch-v1.0": SnapshotBranch( target_type=TargetType.REVISION, target=rev1_swhid.object_id ), b"branch-v2.0": SnapshotBranch( target_type=TargetType.REVISION, target=rev4_swhid.object_id ), b"branch-v3.0": SnapshotBranch( target_type=TargetType.REVISION, target=rev4_swhid.object_id ), b"branch-v4.0": SnapshotBranch( target_type=TargetType.REVISION, target=rev4_swhid.object_id ), } ) > assert snapshot_get_latest(storage, origin) == snapshot E AssertionError: assert Snapshot(bran...x03\x98\x95 ') == Snapshot(bran...O i\x82\x020') E E Differing attributes: E ['branches', 'id'] E E Drill down into differing attribute branches: E branches: ImmutableDict({b'v1.0': SnapshotBranch(target=b'aaaaaaaaaaaaaaaaaaaa', target_type=<TargetType.REVISION: 'revision'>), b'v2.0': SnapshotBranch(target=b'bbbbbbbbbbbbbbbbbbbb', target_type=<TargetType.REVISION: 'revision'>), b'v3.0': SnapshotBranch(target=b'cccccccccccccccccccc', target_type=<TargetType.REVISION: 'revision'>)}) != ImmutableDict({b'branch-v1.0': SnapshotBranch(target=b'aaaaaaaaaaaaaaaaaaaa', target_type=<TargetType.REVISION: 'revision'>), b'branch-v2.0': Snapsho... E E ...Full output truncated (13 lines hidden), use '-vv' to show .tox/py3/lib/python3.7/site-packages/swh/loader/package/tests/test_loader.py:252: AssertionError