Differential D8023 Diff 29047 swh/provenance/tests/test_cli.py

Changeset View

Standalone View

swh/provenance/tests/test_cli.py

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 2021-2022 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

from typing import Set		from typing import Dict, List, Set

from _pytest.monkeypatch import MonkeyPatch		from _pytest.monkeypatch import MonkeyPatch
from click.testing import CliRunner		from click.testing import CliRunner
import psycopg2.extensions		import psycopg2.extensions
import pytest		import pytest

from swh.core.cli import swh as swhmain		from swh.core.cli import swh as swhmain
import swh.core.cli.db # noqa ; ensure cli is loaded		import swh.core.cli.db # noqa ; ensure cli is loaded
from swh.core.db import BaseDb		from swh.core.db import BaseDb
from swh.core.db.db_utils import init_admin_extensions		from swh.core.db.db_utils import init_admin_extensions
		from swh.model.hashutil import MultiHash
import swh.provenance.cli # noqa ; ensure cli is loaded		import swh.provenance.cli # noqa ; ensure cli is loaded
		from swh.provenance.tests.conftest import fill_storage, load_repo_data
		from swh.storage.interface import StorageInterface

		from .conftest import get_datafile
		from .test_utils import invoke, write_configuration_path


def test_cli_swh_db_help() -> None:		def test_cli_swh_db_help() -> None:
# swhmain.add_command(provenance_cli)		# swhmain.add_command(provenance_cli)
result = CliRunner().invoke(swhmain, ["provenance", "-h"])		result = CliRunner().invoke(swhmain, ["provenance", "-h"])
assert result.exit_code == 0		assert result.exit_code == 0
assert "Commands:" in result.output		assert "Commands:" in result.output
commands = result.output.split("Commands:")[1]		commands = result.output.split("Commands:")[1]
▲ Show 20 Lines • Show All 76 Lines • ▼ Show 20 Lines	def test_cli_init_db_default_flavor(postgresql: psycopg2.extensions.connection) -> None:
dbname = postgresql.dsn		dbname = postgresql.dsn
init_admin_extensions("swh.provenance", dbname)		init_admin_extensions("swh.provenance", dbname)
result = CliRunner().invoke(swhmain, ["db", "init", "-d", dbname, "provenance"])		result = CliRunner().invoke(swhmain, ["db", "init", "-d", dbname, "provenance"])
assert result.exit_code == 0, result.output		assert result.exit_code == 0, result.output

with postgresql.cursor() as cur:		with postgresql.cursor() as cur:
cur.execute("select swh_get_dbflavor()")		cur.execute("select swh_get_dbflavor()")
assert cur.fetchone() == ("with-path",)		assert cur.fetchone() == ("with-path",)


		@pytest.mark.parametrize(
		"subcommand",
		(["origin", "from-csv"], ["iter-origins"]),
		)
		def test_cli_origin_from_csv(
		swh_storage: StorageInterface,
		subcommand: List[str],
		swh_storage_backend_config: Dict,
		provenance,
		tmp_path,
		):
		repo = "cmdbts2"
		origin_url = f"https://{repo}"
		data = load_repo_data(repo)
		fill_storage(swh_storage, data)

		assert len(data["origin"]) == 1
		assert {"url": origin_url} in data["origin"]

		cfg = {
		"provenance": {
		"archive": {
		"cls": "api",
		"storage": swh_storage_backend_config,
		},
		"storage": {
		"cls": "postgresql",
		# "db": provenance.storage.conn.dsn,
		"db": provenance.storage.conn.get_dsn_parameters(),
		},
		},
		}

		config_path = write_configuration_path(cfg, tmp_path)

		csv_filepath = get_datafile("origins.csv")
		subcommand = subcommand + [csv_filepath]

		result = invoke(subcommand, config_path)
		assert result.exit_code == 0, f"Unexpected result: {result.output}"

		origin_sha1 = MultiHash.from_data(
		origin_url.encode(), hash_names=["sha1"]
		).digest()["sha1"]
		actual_result = provenance.storage.origin_get([origin_sha1])

		assert actual_result == {origin_sha1: origin_url}