swh_config = '/tmp/pytest-of-jenkins/pytest-0/test_eoferror0/loader.yml' requests_mock_datadir = <requests_mock.mocker.Mocker object at 0x7fad82049a58> def test_eoferror(swh_config, requests_mock_datadir): """Load a truncated archive which is invalid to make the uncompress function raising the exception EOFError. We then check if a snapshot is created, meaning this error is well managed. """ sources = ( "https://nix-community.github.io/nixpkgs-swh/sources-EOFError.json" # noqa ) loader = NixGuixLoader(sources) loader.load() expected_branches = { "evaluation": { "target": "cc4e04c26672dd74e5fd0fecb78b435fb55368f7", "target_type": "revision", }, } expected_snapshot = { "id": "4257fa2350168c6bfec726a06452ea27a2c0cb33", "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:370: _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ snapshot = Snapshot(branches=<swh.model.collections.ImmutableDict object at 0x7fad821dfd68>, id=b"BW\xfa#P\x16\x8ck\xfe\xc7&\xa0dR\xea'\xa2\xc0\xcb3") storage = <swh.storage.retry.RetryingProxyStorage object at 0x7fad82049dd8> 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