Page MenuHomeSoftware Heritage
Differential D4381 Diff 15493 swh/core/db/tests/test_cli.py

Changeset View

Standalone View

View Options

swh/core/db/tests/test_cli.py

# # Copyright (C) 2019-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import copy
import glob
from os import path
from click.testing import CliRunner from click.testing import CliRunner
import pytest
from swh.core.cli.db import db as swhdb from swh.core.cli.db import db as swhdb
from swh.core.db import BaseDb
from swh.core.db.pytest_plugin import postgresql_fact
@pytest.fixture
def cli_runner():
return CliRunner()
help_msg = """Usage: swh [OPTIONS] COMMAND [ARGS]... help_msg = """Usage: swh [OPTIONS] COMMAND [ARGS]...
Command line interface for Software Heritage. Command line interface for Software Heritage.
Options: Options:
-l, --log-level [NOTSET|DEBUG|INFO|WARNING|ERROR|CRITICAL] -l, --log-level [NOTSET|DEBUG|INFO|WARNING|ERROR|CRITICAL]
Log level (defaults to INFO). Log level (defaults to INFO).
--log-config FILENAME Python yaml logging configuration file. --log-config FILENAME Python yaml logging configuration file.
--sentry-dsn TEXT DSN of the Sentry instance to report to --sentry-dsn TEXT DSN of the Sentry instance to report to
-h, --help Show this message and exit. -h, --help Show this message and exit.
Notes: Notes:
If both options are present, --log-level will override the root logger If both options are present, --log-level will override the root logger
configuration set in --log-config. configuration set in --log-config.
The --log-config YAML must conform to the logging.config.dictConfig schema The --log-config YAML must conform to the logging.config.dictConfig schema
documented at https://docs.python.org/3/library/logging.config.html. documented at https://docs.python.org/3/library/logging.config.html.
Commands: Commands:
db Software Heritage database generic tools. db Software Heritage database generic tools.
""" """
def test_swh_help(swhmain): def test_cli_swh_help(swhmain, cli_runner):
swhmain.add_command(swhdb) swhmain.add_command(swhdb)
runner = CliRunner() result = cli_runner.invoke(swhmain, ["-h"])
result = runner.invoke(swhmain, ["-h"])
assert result.exit_code == 0 assert result.exit_code == 0
assert result.output == help_msg assert result.output == help_msg
help_db_msg = """Usage: swh db [OPTIONS] COMMAND [ARGS]... help_db_msg = """Usage: swh db [OPTIONS] COMMAND [ARGS]...
Software Heritage database generic tools. Software Heritage database generic tools.
Options: Options:
-C, --config-file FILE Configuration file. -C, --config-file FILE Configuration file.
-h, --help Show this message and exit. -h, --help Show this message and exit.
Commands: Commands:
create Create a database for the Software Heritage <module>. create Create a database for the Software Heritage <module>.
init Initialize a database for the Software Heritage <module>. init Initialize a database for the Software Heritage <module>.
init-admin Execute superuser-level initialization steps (e.g pg init-admin Execute superuser-level initialization steps (e.g pg
extensions,... extensions,...
""" """
def test_swh_db_help(swhmain): def test_cli_swh_db_help(swhmain, cli_runner):
swhmain.add_command(swhdb) swhmain.add_command(swhdb)
runner = CliRunner() result = cli_runner.invoke(swhmain, ["db", "-h"])
result = runner.invoke(swhmain, ["db", "-h"])
assert result.exit_code == 0 assert result.exit_code == 0
assert result.output == help_db_msg assert result.output == help_db_msg
@pytest.fixture()
def mock_package_sql(mocker, datadir):
"""This bypasses the module manipulation to only returns the data test files.
"""
from swh.core.utils import numfile_sortkey as sortkey
mock_sql_files = mocker.patch("swh.core.cli.db.get_sql_for_package")
sql_files = sorted(glob.glob(path.join(datadir, "cli", "*.sql")), key=sortkey)
mock_sql_files.return_value = sql_files
return mock_sql_files
# We do not want the truncate behavior for those tests
test_db = postgresql_fact(
"postgresql_proc", db_name="clidb", no_truncate_tables={"dbversion", "origin"}
)
@pytest.fixture
def swh_db_cli(cli_runner, monkeypatch, test_db):
"""This initializes a cli_runner and sets the correct environment variable expected by
the cli to run appropriately (when not specifying the --db-name flag)
"""
db_params = test_db.get_dsn_parameters()
monkeypatch.setenv("PGHOST", db_params["host"])
monkeypatch.setenv("PGUSER", db_params["user"])
monkeypatch.setenv("PGPORT", db_params["port"])
return cli_runner, db_params
def craft_conninfo(test_db, dbname=None) -> str:
"""Craft conninfo string out of the test_db object. This also allows to override the
dbname."""
db_params = test_db.get_dsn_parameters()
if dbname:
params = copy.deepcopy(db_params)
params["dbname"] = dbname
else:
params = db_params
return "postgresql://{user}@{host}:{port}/{dbname}".format(**params)
def test_cli_swh_db_create_and_init_db(cli_runner, test_db, mock_package_sql):
"""Create a db then initializing it should be ok
"""
module_name = "something"
conninfo = craft_conninfo(test_db, "new-db")
# This creates the db and installs the necessary admin extensions
result = cli_runner.invoke(swhdb, ["create", module_name, "--db-name", conninfo])
assert result.exit_code == 0, f"Unexpected output: {result.output}"
# This initializes the schema and data
result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo])
assert result.exit_code == 0, f"Unexpected output: {result.output}"
# the origin value in the scripts uses a hash function (which implementation wise
# uses a function from the pgcrypt extension, installed during db creation step)
with BaseDb.connect(conninfo).cursor() as cur:
cur.execute("select * from origin")
origins = cur.fetchall()
assert len(origins) == 1
def test_cli_swh_db_initialization_fail_without_creation_first(
cli_runner, test_db, mock_package_sql
):
"""Init command on an inexisting db cannot work
"""
module_name = "anything" # it's mocked here
conninfo = craft_conninfo(test_db, "inexisting-db")
result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo])
# Fails because we cannot connect to an inexisting db
assert result.exit_code == 1, f"Unexpected output: {result.output}"
def test_cli_swh_db_initialization_fail_without_extension(
cli_runner, test_db, mock_package_sql
):
"""Init command cannot work without privileged extension.
In this test, the schema needs privileged extension to work.
"""
module_name = "anything" # it's mocked here
conninfo = craft_conninfo(test_db)
result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo])
# Fails as the function `public.digest` is not installed, init-admin calls is needed
# first (the next tests show such behavior)
assert result.exit_code == 1, f"Unexpected output: {result.output}"
def test_cli_swh_db_initialization_works_with_flags(
cli_runner, test_db, mock_package_sql
):
"""Init commands with carefully crafted libpq conninfo works
"""
module_name = "anything" # it's mocked here
conninfo = craft_conninfo(test_db)
result = cli_runner.invoke(
swhdb, ["init-admin", module_name, "--db-name", conninfo]
)
assert result.exit_code == 0, f"Unexpected output: {result.output}"
result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo])
assert result.exit_code == 0, f"Unexpected output: {result.output}"
# the origin values in the scripts uses a hash function (which implementation wise
# uses a function from the pgcrypt extension, init-admin calls installs it)
with BaseDb.connect(test_db.dsn).cursor() as cur:
cur.execute("select * from origin")
origins = cur.fetchall()
assert len(origins) == 1
def test_cli_swh_db_initialization_with_env(swh_db_cli, mock_package_sql, test_db):
"""Init commands with standard environment variables works
"""
module_name = "anything" # it's mocked here
cli_runner, db_params = swh_db_cli
result = cli_runner.invoke(
swhdb, ["init-admin", module_name, "--db-name", db_params["dbname"]]
)
assert result.exit_code == 0, f"Unexpected output: {result.output}"
result = cli_runner.invoke(
swhdb, ["init", module_name, "--db-name", db_params["dbname"]]
)
assert result.exit_code == 0, f"Unexpected output: {result.output}"
# the origin values in the scripts uses a hash function (which implementation wise
# uses a function from the pgcrypt extension, init-admin calls installs it)
with BaseDb.connect(test_db.dsn).cursor() as cur:
cur.execute("select * from origin")
origins = cur.fetchall()
assert len(origins) == 1
def test_cli_swh_db_initialization_idempotent(swh_db_cli, mock_package_sql, test_db):
"""Multiple runs of the init commands are idempotent
"""
module_name = "anything" # mocked
cli_runner, db_params = swh_db_cli
result = cli_runner.invoke(
swhdb, ["init-admin", module_name, "--db-name", db_params["dbname"]]
)
assert result.exit_code == 0, f"Unexpected output: {result.output}"
result = cli_runner.invoke(
swhdb, ["init", module_name, "--db-name", db_params["dbname"]]
)
assert result.exit_code == 0, f"Unexpected output: {result.output}"
result = cli_runner.invoke(
swhdb, ["init-admin", module_name, "--db-name", db_params["dbname"]]
)
assert result.exit_code == 0, f"Unexpected output: {result.output}"
result = cli_runner.invoke(
swhdb, ["init", module_name, "--db-name", db_params["dbname"]]
)
assert result.exit_code == 0, f"Unexpected output: {result.output}"
# the origin values in the scripts uses a hash function (which implementation wise
# uses a function from the pgcrypt extension, init-admin calls installs it)
with BaseDb.connect(test_db.dsn).cursor() as cur:
cur.execute("select * from origin")
origins = cur.fetchall()
assert len(origins) == 1
About · Developer information · Legal