swh_config = '/tmp/pytest-of-jenkins/pytest-0/test_raise_exception0/loader.yml' requests_mock_datadir = <requests_mock.mocker.Mocker object at 0x7fad8219dac8> mocker = <pytest_mock.plugin.MockFixture object at 0x7fad821df940> def test_raise_exception(swh_config, requests_mock_datadir, mocker): mock_download = mocker.patch("swh.loader.package.loader.download") mock_download.side_effect = fake_download loader = NixGuixLoader(sources_url) res = loader.load() expected_snapshot_id = "0c5881c74283793ebe9a09a105a9381e41380383" assert res == { "status": "eventful", "snapshot_id": expected_snapshot_id, } expected_branches = { "https://github.com/owner-1/repository-1/revision-1.tgz": { "target": "488ad4e7b8e2511258725063cf43a2b897c503b4", "target_type": "revision", }, "evaluation": { "target": "cc4e04c26672dd74e5fd0fecb78b435fb55368f7", "target_type": "revision", }, } expected_snapshot = { "id": expected_snapshot_id, "branches": expected_branches, } > snapshot = check_snapshot(expected_snapshot, storage=loader.storage) .tox/py3/lib/python3.7/site-packages/swh/loader/package/nixguix/tests/test_nixguix.py:422: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ snapshot = Snapshot(branches=<swh.model.collections.ImmutableDict object at 0x7fad82020fd0>, id=b'\x0cX\x81\xc7B\x83y>\xbe\x9a\t\xa1\x05\xa98\x1eA8\x03\x83') storage = <swh.storage.retry.RetryingProxyStorage object at 0x7fad82054cf8> def check_snapshot( snapshot: Union[Dict[str, Any], Snapshot], storage: StorageInterface ): """Check that: - snapshot exists in the storage and match - each object reference up to the revision/release targets exists Args: snapshot: full snapshot to check for existence and consistency storage: storage to lookup information into Returns: the snapshot stored in the storage for further test assertion if any is needed. """ if isinstance(snapshot, Snapshot): expected_snapshot = snapshot.to_dict() elif isinstance(snapshot, dict): expected_snapshot = snapshot else: raise AssertionError(f"variable 'snapshot' must be a snapshot: {snapshot!r}") expected_snapshot_id = expected_snapshot["id"] expected_branches = expected_snapshot["branches"] snap = storage.snapshot_get(hash_to_bytes(expected_snapshot_id)) if snap is None: raise AssertionError(f"Snapshot {expected_snapshot_id} is not found") snap.pop("next_branch") snapshot = Snapshot.from_dict(snap) assert isinstance(snapshot, Snapshot) objects = defaultdict(list) # recursively check the objects referenced by the snapshots exists expected_branches = {} for branch, target in expected_snapshot["branches"].items(): if isinstance(branch, bytes): branch = branch.decode("utf-8") expected_branches[branch] = decode_target(target) snapshot_branches = {} for branch, target in snap["branches"].items(): if isinstance(branch, bytes): branch = branch.decode("utf-8") snapshot_branches[branch] = decode_target(target) target_type = target["target_type"] objects[target_type].append(target["target"]) # Check snapshot is compliant regarding expectations assert expected_branches == snapshot_branches # Now check for more consistency aliases: List[bytes] = objects.get("alias", []) for alias in aliases: # ensure the snapshot alias target consistent references if alias not in snapshot.branches: raise InconsistentAliasBranchError( f"Alias branch {alias.decode('utf-8')} " f"should be in {list(snapshot.branches)}" ) revs = objects.get("revision") if revs: revisions = list(storage.revision_get(revs)) not_found = [rev_id for rev_id, rev in zip(revs, revisions) if rev is None] if not_found: raise InexistentObjectsError( > f"Revision(s) {','.join(r.hex() for r in not_found)}" " should exist in storage" ) E swh.loader.tests.InexistentObjectsError: Revision(s) cc4e04c26672dd74e5fd0fecb78b435fb55368f7 should exist in storage .tox/py3/lib/python3.7/site-packages/swh/loader/tests/__init__.py:189: InexistentObjectsError