Skip to content
Snippets Groups Projects

pytest_plugin: Migrate to new fixtures from swh-storage pytest plugin

4 files
+ 52
54
Compare changes
  • Side-by-side
  • Inline
Files
4
@@ -167,15 +167,6 @@ SNAPSHOT = Snapshot(
)
@pytest.fixture
def swh_storage_backend_config(swh_storage_postgresql):
return {
"cls": "postgresql",
"db": swh_storage_postgresql.info.dsn,
"objstorage": {"cls": "memory"},
}
@pytest.fixture
def mock_storage(mocker):
mock_storage = mocker.patch("swh.loader.tests.origin_get_latest_visit_status")
@@ -315,7 +306,7 @@ def test_encode_target():
}
def test_check_snapshot(swh_storage):
def test_check_snapshot(swh_storage_backend):
"""Everything should be fine when snapshot is found and the snapshot reference up to the
revision exist in the storage.
@@ -340,11 +331,11 @@ def test_check_snapshot(swh_storage):
else:
assert target.target in [REVISION.id, RELEASE.id]
swh_storage.content_add([CONTENT])
swh_storage.directory_add([DIRECTORY])
swh_storage.revision_add([REVISION])
swh_storage.release_add([RELEASE])
s = swh_storage.snapshot_add([SNAPSHOT])
swh_storage_backend.content_add([CONTENT])
swh_storage_backend.directory_add([DIRECTORY])
swh_storage_backend.revision_add([REVISION])
swh_storage_backend.release_add([RELEASE])
s = swh_storage_backend.snapshot_add([SNAPSHOT])
assert s == {
"snapshot:add": 1,
}
@@ -352,12 +343,12 @@ def test_check_snapshot(swh_storage):
# all should be fine!
check_snapshot(
SNAPSHOT,
swh_storage,
swh_storage_backend,
allowed_empty=[(SnapshotTargetType.REVISION, b"evaluation")],
)
def test_check_snapshot_failures(swh_storage):
def test_check_snapshot_failures(swh_storage_backend):
"""Failure scenarios:
0. snapshot parameter is not a snapshot
@@ -383,7 +374,7 @@ def test_check_snapshot_failures(swh_storage):
},
)
s = swh_storage.snapshot_add([snapshot])
s = swh_storage_backend.snapshot_add([snapshot])
assert s == {
"snapshot:add": 1,
}
@@ -400,16 +391,20 @@ def test_check_snapshot_failures(swh_storage):
with pytest.raises(
AssertionError, match="argument 'expected_snapshot' must be a snapshot"
):
check_snapshot(ORIGIN_VISIT, swh_storage)
check_snapshot(ORIGIN_VISIT, swh_storage_backend)
# 1. snapshot id is correct but branches mismatched
with pytest.raises(AssertionError): # sadly debian build raises only assertion
check_snapshot(attr.evolve(unexpected_snapshot, id=snapshot.id), swh_storage)
check_snapshot(
attr.evolve(unexpected_snapshot, id=snapshot.id), swh_storage_backend
)
# 2. snapshot id is not correct, it's not found in the storage
wrong_snap_id = hash_to_bytes("999666f535f882bc7f9a18fb16c9ad27fda7bab7")
with pytest.raises(AssertionError, match="is not found"):
check_snapshot(attr.evolve(unexpected_snapshot, id=wrong_snap_id), swh_storage)
check_snapshot(
attr.evolve(unexpected_snapshot, id=wrong_snap_id), swh_storage_backend
)
# 3. snapshot references an inexistent alias
snapshot0 = Snapshot(
@@ -421,14 +416,14 @@ def test_check_snapshot_failures(swh_storage):
),
},
)
swh_storage.snapshot_add([snapshot0])
swh_storage_backend.snapshot_add([snapshot0])
with pytest.raises(InconsistentAliasBranchError, match="Alias branch HEAD"):
check_snapshot(snapshot0, swh_storage)
check_snapshot(snapshot0, swh_storage_backend)
# 4. snapshot is found in storage, targeted revision does not exist
rev_not_found = list(swh_storage.revision_missing([REVISION.id]))
rev_not_found = list(swh_storage_backend.revision_missing([REVISION.id]))
assert len(rev_not_found) == 1
snapshot1 = Snapshot(
@@ -445,17 +440,17 @@ def test_check_snapshot_failures(swh_storage):
},
)
swh_storage.snapshot_add([snapshot1])
swh_storage_backend.snapshot_add([snapshot1])
with pytest.raises(InexistentObjectsError, match="Branch/Revision"):
check_snapshot(snapshot1, swh_storage)
check_snapshot(snapshot1, swh_storage_backend)
# 5. snapshot is found in storage, targeted revision exists but the directory the
# revision targets does not exist
swh_storage.revision_add([REVISION])
swh_storage_backend.revision_add([REVISION])
dir_not_found = list(swh_storage.directory_missing([REVISION.directory]))
dir_not_found = list(swh_storage_backend.directory_missing([REVISION.directory]))
assert len(dir_not_found) == 1
snapshot2 = Snapshot(
@@ -472,21 +467,23 @@ def test_check_snapshot_failures(swh_storage):
},
)
swh_storage.snapshot_add([snapshot2])
swh_storage_backend.snapshot_add([snapshot2])
with pytest.raises(InexistentObjectsError, match="Missing directories"):
check_snapshot(snapshot2, swh_storage)
check_snapshot(snapshot2, swh_storage_backend)
assert DIRECTORY.id == REVISION.directory
swh_storage.directory_add([DIRECTORY])
swh_storage_backend.directory_add([DIRECTORY])
# 6. snapshot is found in storage, target revision exists, targeted directory by the
# revision exist. Content targeted by the directory does not exist.
assert DIRECTORY.entries[0].target == CONTENT.sha1_git
not_found = list(swh_storage.content_missing_per_sha1_git([CONTENT.sha1_git]))
not_found = list(
swh_storage_backend.content_missing_per_sha1_git([CONTENT.sha1_git])
)
assert len(not_found) == 1
swh_storage.directory_add([DIRECTORY])
swh_storage_backend.directory_add([DIRECTORY])
snapshot3 = Snapshot(
id=hash_to_bytes("091456f535f882bc7f9a18fb16c9ad27fda7bab7"),
@@ -502,9 +499,9 @@ def test_check_snapshot_failures(swh_storage):
},
)
swh_storage.snapshot_add([snapshot3])
swh_storage_backend.snapshot_add([snapshot3])
with pytest.raises(InexistentObjectsError, match="Missing content(s)"):
check_snapshot(snapshot3, swh_storage)
check_snapshot(snapshot3, swh_storage_backend)
# 7. snapshot is found in storage, targeted release does not exist
@@ -529,7 +526,7 @@ def test_check_snapshot_failures(swh_storage):
},
)
swh_storage.snapshot_add([snapshot4])
swh_storage_backend.snapshot_add([snapshot4])
with pytest.raises(InexistentObjectsError, match="Branch/Release"):
check_snapshot(snapshot4, swh_storage)
check_snapshot(snapshot4, swh_storage_backend)
Loading