Page MenuHomeSoftware Heritage
Differential D3383 Diff 12007 swh/storage/tests/test_storage.py

Changeset View

Standalone View

View Options

swh/storage/tests/test_storage.py

Show First 20 Lines • Show All 3,432 Lines • ▼ Show 20 Lines def test_content_metadata_get_paginate_same_date(self, swh_storage):
assert [data.content_metadata] == result["results"] assert [data.content_metadata] == result["results"]
result = swh_storage.content_metadata_get( result = swh_storage.content_metadata_get(
content_swhid, authority, limit=1, page_token=result["next_page_token"] content_swhid, authority, limit=1, page_token=result["next_page_token"]
) )
assert result["next_page_token"] is None assert result["next_page_token"] is None
assert [content_metadata2] == result["results"] assert [content_metadata2] == result["results"]
def test_directory_metadata_add(self, swh_storage):
directory = data.dir
fetcher = data.metadata_fetcher
authority = data.metadata_authority
directory_swhid = f"swh:1:dir:{directory['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.directory_metadata_add(**data.directory_metadata)
swh_storage.directory_metadata_add(**data.directory_metadata2)
result = swh_storage.directory_metadata_get(directory_swhid, authority)
assert result["next_page_token"] is None
assert [data.directory_metadata, data.directory_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
def test_directory_metadata_add_duplicate(self, swh_storage):
"""Duplicates should be silently updated."""
directory = data.dir
fetcher = data.metadata_fetcher
authority = data.metadata_authority
directory_swhid = f"swh:1:dir:{directory['id']}"
new_directory_metadata2 = {
**data.directory_metadata2,
"format": "new-format",
"metadata": b"new-metadata",
}
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.directory_metadata_add(**data.directory_metadata)
swh_storage.directory_metadata_add(**data.directory_metadata2)
swh_storage.directory_metadata_add(**new_directory_metadata2)
result = swh_storage.directory_metadata_get(directory_swhid, authority)
assert result["next_page_token"] is None
expected_results1 = (data.directory_metadata, new_directory_metadata2)
expected_results2 = (data.directory_metadata, data.directory_metadata2)
assert tuple(sorted(result["results"], key=lambda x: x["discovery_date"],)) in (
expected_results1, # cassandra
expected_results2, # postgresql
)
def test_directory_metadata_add_dict(self, swh_storage):
fetcher = data.metadata_fetcher
authority = data.metadata_authority
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
kwargs = data.directory_metadata.copy()
kwargs["metadata"] = {"foo": "bar"}
with pytest.raises(StorageArgumentException):
swh_storage.directory_metadata_add(**kwargs)
def test_directory_metadata_get(self, swh_storage):
authority = data.metadata_authority
fetcher = data.metadata_fetcher
authority2 = data.metadata_authority2
fetcher2 = data.metadata_fetcher2
directory1_swhid = f"swh:1:dir:{data.dir['id']}"
directory2_swhid = f"swh:1:dir:{data.dir2['id']}"
directory1_metadata1 = data.directory_metadata
directory1_metadata2 = data.directory_metadata2
directory1_metadata3 = data.directory_metadata3
directory2_metadata = {**data.directory_metadata2, "id": directory2_swhid}
swh_storage.metadata_authority_add(**authority)
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority2)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.directory_metadata_add(**directory1_metadata1)
swh_storage.directory_metadata_add(**directory1_metadata2)
swh_storage.directory_metadata_add(**directory1_metadata3)
swh_storage.directory_metadata_add(**directory2_metadata)
result = swh_storage.directory_metadata_get(directory1_swhid, authority)
assert result["next_page_token"] is None
assert [directory1_metadata1, directory1_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.directory_metadata_get(directory1_swhid, authority2)
assert result["next_page_token"] is None
assert [directory1_metadata3] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.directory_metadata_get(directory2_swhid, authority)
assert result["next_page_token"] is None
assert [directory2_metadata] == list(result["results"],)
def test_directory_metadata_get_after(self, swh_storage):
directory = data.dir
fetcher = data.metadata_fetcher
authority = data.metadata_authority
directory_swhid = f"swh:1:dir:{directory['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.directory_metadata_add(**data.directory_metadata)
swh_storage.directory_metadata_add(**data.directory_metadata2)
result = swh_storage.directory_metadata_get(
directory_swhid,
authority,
after=data.directory_metadata["discovery_date"] - timedelta(seconds=1),
)
assert result["next_page_token"] is None
assert [data.directory_metadata, data.directory_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.directory_metadata_get(
directory_swhid, authority, after=data.directory_metadata["discovery_date"]
)
assert result["next_page_token"] is None
assert [data.directory_metadata2] == result["results"]
result = swh_storage.directory_metadata_get(
directory_swhid, authority, after=data.directory_metadata2["discovery_date"]
)
assert result["next_page_token"] is None
assert [] == result["results"]
def test_directory_metadata_get_paginate(self, swh_storage):
directory = data.dir
fetcher = data.metadata_fetcher
authority = data.metadata_authority
directory_swhid = f"swh:1:dir:{directory['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.directory_metadata_add(**data.directory_metadata)
swh_storage.directory_metadata_add(**data.directory_metadata2)
swh_storage.directory_metadata_get(directory_swhid, authority)
result = swh_storage.directory_metadata_get(directory_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.directory_metadata] == result["results"]
result = swh_storage.directory_metadata_get(
directory_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [data.directory_metadata2] == result["results"]
def test_directory_metadata_get_paginate_same_date(self, swh_storage):
directory = data.dir
fetcher1 = data.metadata_fetcher
fetcher2 = data.metadata_fetcher2
authority = data.metadata_authority
directory_swhid = f"swh:1:dir:{directory['id']}"
swh_storage.metadata_fetcher_add(**fetcher1)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.metadata_authority_add(**authority)
directory_metadata2 = {
**data.directory_metadata2,
"discovery_date": data.directory_metadata2["discovery_date"],
"fetcher": {"name": fetcher2["name"], "version": fetcher2["version"],},
}
swh_storage.directory_metadata_add(**data.directory_metadata)
swh_storage.directory_metadata_add(**directory_metadata2)
result = swh_storage.directory_metadata_get(directory_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.directory_metadata] == result["results"]
result = swh_storage.directory_metadata_get(
directory_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [directory_metadata2] == result["results"]
def test_revision_metadata_add(self, swh_storage):
revision = data.revision
fetcher = data.metadata_fetcher
authority = data.metadata_authority
revision_swhid = f"swh:1:rev:{revision['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.revision_metadata_add(**data.revision_metadata)
swh_storage.revision_metadata_add(**data.revision_metadata2)
result = swh_storage.revision_metadata_get(revision_swhid, authority)
assert result["next_page_token"] is None
assert [data.revision_metadata, data.revision_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
def test_revision_metadata_add_duplicate(self, swh_storage):
"""Duplicates should be silently updated."""
revision = data.revision
fetcher = data.metadata_fetcher
authority = data.metadata_authority
revision_swhid = f"swh:1:rev:{revision['id']}"
new_revision_metadata2 = {
**data.revision_metadata2,
"format": "new-format",
"metadata": b"new-metadata",
}
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.revision_metadata_add(**data.revision_metadata)
swh_storage.revision_metadata_add(**data.revision_metadata2)
swh_storage.revision_metadata_add(**new_revision_metadata2)
result = swh_storage.revision_metadata_get(revision_swhid, authority)
assert result["next_page_token"] is None
expected_results1 = (data.revision_metadata, new_revision_metadata2)
expected_results2 = (data.revision_metadata, data.revision_metadata2)
assert tuple(sorted(result["results"], key=lambda x: x["discovery_date"],)) in (
expected_results1, # cassandra
expected_results2, # postgresql
)
def test_revision_metadata_add_dict(self, swh_storage):
fetcher = data.metadata_fetcher
authority = data.metadata_authority
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
kwargs = data.revision_metadata.copy()
kwargs["metadata"] = {"foo": "bar"}
with pytest.raises(StorageArgumentException):
swh_storage.revision_metadata_add(**kwargs)
def test_revision_metadata_get(self, swh_storage):
authority = data.metadata_authority
fetcher = data.metadata_fetcher
authority2 = data.metadata_authority2
fetcher2 = data.metadata_fetcher2
revision1_swhid = f"swh:1:rev:{data.revision['id']}"
revision2_swhid = f"swh:1:rev:{data.revision2['id']}"
revision1_metadata1 = data.revision_metadata
revision1_metadata2 = data.revision_metadata2
revision1_metadata3 = data.revision_metadata3
revision2_metadata = {**data.revision_metadata2, "id": revision2_swhid}
swh_storage.metadata_authority_add(**authority)
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority2)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.revision_metadata_add(**revision1_metadata1)
swh_storage.revision_metadata_add(**revision1_metadata2)
swh_storage.revision_metadata_add(**revision1_metadata3)
swh_storage.revision_metadata_add(**revision2_metadata)
result = swh_storage.revision_metadata_get(revision1_swhid, authority)
assert result["next_page_token"] is None
assert [revision1_metadata1, revision1_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.revision_metadata_get(revision1_swhid, authority2)
assert result["next_page_token"] is None
assert [revision1_metadata3] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.revision_metadata_get(revision2_swhid, authority)
assert result["next_page_token"] is None
assert [revision2_metadata] == list(result["results"],)
def test_revision_metadata_get_after(self, swh_storage):
revision = data.revision
fetcher = data.metadata_fetcher
authority = data.metadata_authority
revision_swhid = f"swh:1:rev:{revision['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.revision_metadata_add(**data.revision_metadata)
swh_storage.revision_metadata_add(**data.revision_metadata2)
result = swh_storage.revision_metadata_get(
revision_swhid,
authority,
after=data.revision_metadata["discovery_date"] - timedelta(seconds=1),
)
assert result["next_page_token"] is None
assert [data.revision_metadata, data.revision_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.revision_metadata_get(
revision_swhid, authority, after=data.revision_metadata["discovery_date"]
)
assert result["next_page_token"] is None
assert [data.revision_metadata2] == result["results"]
result = swh_storage.revision_metadata_get(
revision_swhid, authority, after=data.revision_metadata2["discovery_date"]
)
assert result["next_page_token"] is None
assert [] == result["results"]
def test_revision_metadata_get_paginate(self, swh_storage):
revision = data.revision
fetcher = data.metadata_fetcher
authority = data.metadata_authority
revision_swhid = f"swh:1:rev:{revision['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.revision_metadata_add(**data.revision_metadata)
swh_storage.revision_metadata_add(**data.revision_metadata2)
swh_storage.revision_metadata_get(revision_swhid, authority)
result = swh_storage.revision_metadata_get(revision_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.revision_metadata] == result["results"]
result = swh_storage.revision_metadata_get(
revision_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [data.revision_metadata2] == result["results"]
def test_revision_metadata_get_paginate_same_date(self, swh_storage):
revision = data.revision
fetcher1 = data.metadata_fetcher
fetcher2 = data.metadata_fetcher2
authority = data.metadata_authority
revision_swhid = f"swh:1:rev:{revision['id']}"
swh_storage.metadata_fetcher_add(**fetcher1)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.metadata_authority_add(**authority)
revision_metadata2 = {
**data.revision_metadata2,
"discovery_date": data.revision_metadata2["discovery_date"],
"fetcher": {"name": fetcher2["name"], "version": fetcher2["version"],},
}
swh_storage.revision_metadata_add(**data.revision_metadata)
swh_storage.revision_metadata_add(**revision_metadata2)
result = swh_storage.revision_metadata_get(revision_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.revision_metadata] == result["results"]
result = swh_storage.revision_metadata_get(
revision_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [revision_metadata2] == result["results"]
def test_release_metadata_add(self, swh_storage):
release = data.release
fetcher = data.metadata_fetcher
authority = data.metadata_authority
release_swhid = f"swh:1:rel:{release['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.release_metadata_add(**data.release_metadata)
swh_storage.release_metadata_add(**data.release_metadata2)
result = swh_storage.release_metadata_get(release_swhid, authority)
assert result["next_page_token"] is None
assert [data.release_metadata, data.release_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
def test_release_metadata_add_duplicate(self, swh_storage):
"""Duplicates should be silently updated."""
release = data.release
fetcher = data.metadata_fetcher
authority = data.metadata_authority
release_swhid = f"swh:1:rel:{release['id']}"
new_release_metadata2 = {
**data.release_metadata2,
"format": "new-format",
"metadata": b"new-metadata",
}
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.release_metadata_add(**data.release_metadata)
swh_storage.release_metadata_add(**data.release_metadata2)
swh_storage.release_metadata_add(**new_release_metadata2)
result = swh_storage.release_metadata_get(release_swhid, authority)
assert result["next_page_token"] is None
expected_results1 = (data.release_metadata, new_release_metadata2)
expected_results2 = (data.release_metadata, data.release_metadata2)
assert tuple(sorted(result["results"], key=lambda x: x["discovery_date"],)) in (
expected_results1, # cassandra
expected_results2, # postgresql
)
def test_release_metadata_add_dict(self, swh_storage):
fetcher = data.metadata_fetcher
authority = data.metadata_authority
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
kwargs = data.release_metadata.copy()
kwargs["metadata"] = {"foo": "bar"}
with pytest.raises(StorageArgumentException):
swh_storage.release_metadata_add(**kwargs)
def test_release_metadata_get(self, swh_storage):
authority = data.metadata_authority
fetcher = data.metadata_fetcher
authority2 = data.metadata_authority2
fetcher2 = data.metadata_fetcher2
release1_swhid = f"swh:1:rel:{data.release['id']}"
release2_swhid = f"swh:1:rel:{data.release2['id']}"
release1_metadata1 = data.release_metadata
release1_metadata2 = data.release_metadata2
release1_metadata3 = data.release_metadata3
release2_metadata = {**data.release_metadata2, "id": release2_swhid}
swh_storage.metadata_authority_add(**authority)
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority2)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.release_metadata_add(**release1_metadata1)
swh_storage.release_metadata_add(**release1_metadata2)
swh_storage.release_metadata_add(**release1_metadata3)
swh_storage.release_metadata_add(**release2_metadata)
result = swh_storage.release_metadata_get(release1_swhid, authority)
assert result["next_page_token"] is None
assert [release1_metadata1, release1_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.release_metadata_get(release1_swhid, authority2)
assert result["next_page_token"] is None
assert [release1_metadata3] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.release_metadata_get(release2_swhid, authority)
assert result["next_page_token"] is None
assert [release2_metadata] == list(result["results"],)
def test_release_metadata_get_after(self, swh_storage):
release = data.release
fetcher = data.metadata_fetcher
authority = data.metadata_authority
release_swhid = f"swh:1:rel:{release['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.release_metadata_add(**data.release_metadata)
swh_storage.release_metadata_add(**data.release_metadata2)
result = swh_storage.release_metadata_get(
release_swhid,
authority,
after=data.release_metadata["discovery_date"] - timedelta(seconds=1),
)
assert result["next_page_token"] is None
assert [data.release_metadata, data.release_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.release_metadata_get(
release_swhid, authority, after=data.release_metadata["discovery_date"]
)
assert result["next_page_token"] is None
assert [data.release_metadata2] == result["results"]
result = swh_storage.release_metadata_get(
release_swhid, authority, after=data.release_metadata2["discovery_date"]
)
assert result["next_page_token"] is None
assert [] == result["results"]
def test_release_metadata_get_paginate(self, swh_storage):
release = data.release
fetcher = data.metadata_fetcher
authority = data.metadata_authority
release_swhid = f"swh:1:rel:{release['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.release_metadata_add(**data.release_metadata)
swh_storage.release_metadata_add(**data.release_metadata2)
swh_storage.release_metadata_get(release_swhid, authority)
result = swh_storage.release_metadata_get(release_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.release_metadata] == result["results"]
result = swh_storage.release_metadata_get(
release_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [data.release_metadata2] == result["results"]
def test_release_metadata_get_paginate_same_date(self, swh_storage):
release = data.release
fetcher1 = data.metadata_fetcher
fetcher2 = data.metadata_fetcher2
authority = data.metadata_authority
release_swhid = f"swh:1:rel:{release['id']}"
swh_storage.metadata_fetcher_add(**fetcher1)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.metadata_authority_add(**authority)
release_metadata2 = {
**data.release_metadata2,
"discovery_date": data.release_metadata2["discovery_date"],
"fetcher": {"name": fetcher2["name"], "version": fetcher2["version"],},
}
swh_storage.release_metadata_add(**data.release_metadata)
swh_storage.release_metadata_add(**release_metadata2)
result = swh_storage.release_metadata_get(release_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.release_metadata] == result["results"]
result = swh_storage.release_metadata_get(
release_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [release_metadata2] == result["results"]
def test_snapshot_metadata_add(self, swh_storage):
snapshot = data.snapshot
fetcher = data.metadata_fetcher
authority = data.metadata_authority
snapshot_swhid = f"swh:1:snp:{snapshot['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata2)
result = swh_storage.snapshot_metadata_get(snapshot_swhid, authority)
assert result["next_page_token"] is None
assert [data.snapshot_metadata, data.snapshot_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
def test_snapshot_metadata_add_duplicate(self, swh_storage):
"""Duplicates should be silently updated."""
snapshot = data.snapshot
fetcher = data.metadata_fetcher
authority = data.metadata_authority
snapshot_swhid = f"swh:1:snp:{snapshot['id']}"
new_snapshot_metadata2 = {
**data.snapshot_metadata2,
"format": "new-format",
"metadata": b"new-metadata",
}
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata2)
swh_storage.snapshot_metadata_add(**new_snapshot_metadata2)
result = swh_storage.snapshot_metadata_get(snapshot_swhid, authority)
assert result["next_page_token"] is None
expected_results1 = (data.snapshot_metadata, new_snapshot_metadata2)
expected_results2 = (data.snapshot_metadata, data.snapshot_metadata2)
assert tuple(sorted(result["results"], key=lambda x: x["discovery_date"],)) in (
expected_results1, # cassandra
expected_results2, # postgresql
)
def test_snapshot_metadata_add_dict(self, swh_storage):
fetcher = data.metadata_fetcher
authority = data.metadata_authority
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
kwargs = data.snapshot_metadata.copy()
kwargs["metadata"] = {"foo": "bar"}
with pytest.raises(StorageArgumentException):
swh_storage.snapshot_metadata_add(**kwargs)
def test_snapshot_metadata_get(self, swh_storage):
authority = data.metadata_authority
fetcher = data.metadata_fetcher
authority2 = data.metadata_authority2
fetcher2 = data.metadata_fetcher2
snapshot1_swhid = f"swh:1:snp:{data.snapshot['id']}"
snapshot2_swhid = f"swh:1:snp:{data.complete_snapshot['id']}"
snapshot1_metadata1 = data.snapshot_metadata
snapshot1_metadata2 = data.snapshot_metadata2
snapshot1_metadata3 = data.snapshot_metadata3
snapshot2_metadata = {**data.snapshot_metadata2, "id": snapshot2_swhid}
swh_storage.metadata_authority_add(**authority)
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority2)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.snapshot_metadata_add(**snapshot1_metadata1)
swh_storage.snapshot_metadata_add(**snapshot1_metadata2)
swh_storage.snapshot_metadata_add(**snapshot1_metadata3)
swh_storage.snapshot_metadata_add(**snapshot2_metadata)
result = swh_storage.snapshot_metadata_get(snapshot1_swhid, authority)
assert result["next_page_token"] is None
assert [snapshot1_metadata1, snapshot1_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.snapshot_metadata_get(snapshot1_swhid, authority2)
assert result["next_page_token"] is None
assert [snapshot1_metadata3] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.snapshot_metadata_get(snapshot2_swhid, authority)
assert result["next_page_token"] is None
assert [snapshot2_metadata] == list(result["results"],)
def test_snapshot_metadata_get_after(self, swh_storage):
snapshot = data.snapshot
fetcher = data.metadata_fetcher
authority = data.metadata_authority
snapshot_swhid = f"swh:1:snp:{snapshot['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata2)
result = swh_storage.snapshot_metadata_get(
snapshot_swhid,
authority,
after=data.snapshot_metadata["discovery_date"] - timedelta(seconds=1),
)
assert result["next_page_token"] is None
assert [data.snapshot_metadata, data.snapshot_metadata2] == list(
sorted(result["results"], key=lambda x: x["discovery_date"],)
)
result = swh_storage.snapshot_metadata_get(
snapshot_swhid, authority, after=data.snapshot_metadata["discovery_date"]
)
assert result["next_page_token"] is None
assert [data.snapshot_metadata2] == result["results"]
result = swh_storage.snapshot_metadata_get(
snapshot_swhid, authority, after=data.snapshot_metadata2["discovery_date"]
)
assert result["next_page_token"] is None
assert [] == result["results"]
def test_snapshot_metadata_get_paginate(self, swh_storage):
snapshot = data.snapshot
fetcher = data.metadata_fetcher
authority = data.metadata_authority
snapshot_swhid = f"swh:1:snp:{snapshot['id']}"
swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata)
swh_storage.snapshot_metadata_add(**data.snapshot_metadata2)
swh_storage.snapshot_metadata_get(snapshot_swhid, authority)
result = swh_storage.snapshot_metadata_get(snapshot_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.snapshot_metadata] == result["results"]
result = swh_storage.snapshot_metadata_get(
snapshot_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [data.snapshot_metadata2] == result["results"]
def test_snapshot_metadata_get_paginate_same_date(self, swh_storage):
snapshot = data.snapshot
fetcher1 = data.metadata_fetcher
fetcher2 = data.metadata_fetcher2
authority = data.metadata_authority
snapshot_swhid = f"swh:1:snp:{snapshot['id']}"
swh_storage.metadata_fetcher_add(**fetcher1)
swh_storage.metadata_fetcher_add(**fetcher2)
swh_storage.metadata_authority_add(**authority)
snapshot_metadata2 = {
**data.snapshot_metadata2,
"discovery_date": data.snapshot_metadata2["discovery_date"],
"fetcher": {"name": fetcher2["name"], "version": fetcher2["version"],},
}
swh_storage.snapshot_metadata_add(**data.snapshot_metadata)
swh_storage.snapshot_metadata_add(**snapshot_metadata2)
result = swh_storage.snapshot_metadata_get(snapshot_swhid, authority, limit=1)
assert result["next_page_token"] is not None
assert [data.snapshot_metadata] == result["results"]
result = swh_storage.snapshot_metadata_get(
snapshot_swhid, authority, limit=1, page_token=result["next_page_token"]
)
assert result["next_page_token"] is None
assert [snapshot_metadata2] == result["results"]
def test_origin_metadata_add(self, swh_storage): def test_origin_metadata_add(self, swh_storage):
origin = data.origin origin = data.origin
fetcher = data.metadata_fetcher fetcher = data.metadata_fetcher
authority = data.metadata_authority authority = data.metadata_authority
assert swh_storage.origin_add([origin]) == {"origin:add": 1} assert swh_storage.origin_add([origin]) == {"origin:add": 1}
swh_storage.metadata_fetcher_add(**fetcher) swh_storage.metadata_fetcher_add(**fetcher)
swh_storage.metadata_authority_add(**authority) swh_storage.metadata_authority_add(**authority)
▲ Show 20 Lines • Show All 717 Lines • Show Last 20 Lines
About · Developer information · Legal