swh_storage = <swh.storage.postgresql.storage.Storage object at 0x7f04f932b6a0>
mismatch_on = 'revision1'
@pytest.mark.parametrize(
"mismatch_on", ["content", "directory", "revision1", "revision2", "none"]
)
def test_checksum_mismatch(swh_storage, mismatch_on):
date = TimestampWithTimezone.from_datetime(
datetime.datetime(2021, 5, 7, 8, 43, 59, tzinfo=datetime.timezone.utc)
)
author = Person.from_fullname(b"Foo <foo@example.org>")
wrong_hash = b"\x12\x34" * 10
cnt1 = Content.from_data(b"Tr0ub4dor&3")
if mismatch_on == "content":
cnt1 = attr.evolve(cnt1, sha1_git=wrong_hash)
dir1 = Directory(
entries=(
DirectoryEntry(
name=b"file1",
type="file",
perms=DentryPerms.content,
target=cnt1.sha1_git,
),
)
)
if mismatch_on == "directory":
dir1 = attr.evolve(dir1, id=wrong_hash)
rev1 = Revision(
message=b"msg1",
date=date,
committer_date=date,
author=author,
committer=author,
directory=dir1.id,
type=RevisionType.GIT,
synthetic=True,
)
if mismatch_on == "revision1":
rev1 = attr.evolve(rev1, id=wrong_hash)
rev2 = Revision(
message=b"msg2",
date=date,
committer_date=date,
author=author,
committer=author,
directory=dir1.id,
parents=(rev1.id,),
type=RevisionType.GIT,
synthetic=True,
)
if mismatch_on == "revision2":
rev2 = attr.evolve(rev2, id=wrong_hash)
cooked_swhid = rev2.swhid()
swh_storage.content_add([cnt1])
swh_storage.directory_add([dir1])
swh_storage.revision_add([rev1, rev2])
backend = InMemoryVaultBackend()
cooker = GitBareCooker(
cooked_swhid, backend=backend, storage=swh_storage, graph=None,
)
cooker.cook()
# Get bundle
> bundle = backend.fetch("git_bare", cooked_swhid)
.tox/py3/lib/python3.7/site-packages/swh/vault/tests/test_git_bare_cooker.py:413:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
.tox/py3/lib/python3.7/site-packages/swh/vault/in_memory_backend.py:20: in fetch
return self._cache.get(bundle_type, swhid)
.tox/py3/lib/python3.7/site-packages/swh/vault/cache.py:28: in get
return self.objstorage.get(hashutil.hash_to_bytes(sid))
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <swh.objstorage.backends.in_memory.InMemoryObjStorage object at 0x7f04f932b7f0>
obj_id = b'\x8b\x8a8\xa8O\xc6y\xa8)\xb6\xf6\xd03\xd0@\xc6\x91g\xd7K', args = ()
kwargs = {}
def get(self, obj_id, *args, **kwargs):
if obj_id not in self:
> raise ObjNotFoundError(obj_id)
E swh.objstorage.exc.ObjNotFoundError: object not found: b'\x8b\x8a8\xa8O\xc6y\xa8)\xb6\xf6\xd03\xd0@\xc6\x91g\xd7K'
.tox/py3/lib/python3.7/site-packages/swh/objstorage/backends/in_memory.py:46: ObjNotFoundError
TEST RESULT
TEST RESULT
- Run At
- Nov 23 2021, 4:08 PM