Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/tests/test_archive_interface.py
| Show First 20 Lines • Show All 222 Lines • ▼ Show 20 Lines | def test_archive_interface(repo: str, archive: ArchiveInterface) -> None: | ||||
| with BaseDb.connect(dsn).conn as conn: | with BaseDb.connect(dsn).conn as conn: | ||||
| BaseDb.adapt_conn(conn) | BaseDb.adapt_conn(conn) | ||||
| archive_direct = ArchivePostgreSQL(conn) | archive_direct = ArchivePostgreSQL(conn) | ||||
| check_directory_ls(archive, archive_direct, data) | check_directory_ls(archive, archive_direct, data) | ||||
| check_revision_get_some_outbound_edges(archive, archive_direct, data) | check_revision_get_some_outbound_edges(archive, archive_direct, data) | ||||
| check_snapshot_get_heads(archive, archive_direct, data) | check_snapshot_get_heads(archive, archive_direct, data) | ||||
| @pytest.mark.grpc | |||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||
| "repo", | "repo", | ||||
| ("cmdbts2", "out-of-order", "with-merges"), | ("cmdbts2", "out-of-order", "with-merges"), | ||||
| ) | ) | ||||
| def test_archive_graph(repo: str, archive: ArchiveInterface) -> None: | def test_archive_graph(repo: str, archive: ArchiveInterface) -> None: | ||||
| data = load_repo_data(repo) | data = load_repo_data(repo) | ||||
| fill_storage(archive.storage, data) | fill_storage(archive.storage, data) | ||||
| with grpc_server(repo) as url: | with grpc_server(repo) as url: | ||||
| # test against ArchiveGraph | # test against ArchiveGraph | ||||
| archive_graph = ArchiveGraph(url, archive.storage) | archive_graph = ArchiveGraph(url, archive.storage) | ||||
| with pytest.raises(NotImplementedError): | with pytest.raises(NotImplementedError): | ||||
| check_directory_ls(archive, archive_graph, data) | check_directory_ls(archive, archive_graph, data) | ||||
| check_revision_get_some_outbound_edges(archive, archive_graph, data) | check_revision_get_some_outbound_edges(archive, archive_graph, data) | ||||
| check_snapshot_get_heads(archive, archive_graph, data) | check_snapshot_get_heads(archive, archive_graph, data) | ||||
| @pytest.mark.grpc | |||||
| @pytest.mark.parametrize( | @pytest.mark.parametrize( | ||||
| "repo", | "repo", | ||||
| ("cmdbts2", "out-of-order", "with-merges"), | ("cmdbts2", "out-of-order", "with-merges"), | ||||
| ) | ) | ||||
| def test_archive_multiplexed(repo: str, archive: ArchiveInterface) -> None: | def test_archive_multiplexed(repo: str, archive: ArchiveInterface) -> None: | ||||
| # read data/README.md for more details on how these datasets are generated | # read data/README.md for more details on how these datasets are generated | ||||
| data = load_repo_data(repo) | data = load_repo_data(repo) | ||||
| fill_storage(archive.storage, data) | fill_storage(archive.storage, data) | ||||
| Show All 19 Lines | |||||