diff --git a/swh/core/db/tests/data/cli/30-schema.sql b/swh/core/db/tests/data/cli/30-schema.sql index a5f6d2c..fcbd581 100644 --- a/swh/core/db/tests/data/cli/30-schema.sql +++ b/swh/core/db/tests/data/cli/30-schema.sql @@ -1,13 +1,13 @@ -- schema version table which won't get truncated -create table if not exists dbversion ( +create table dbversion ( version int primary key, release timestamptz, description text ); -- origin table create table if not exists origin ( id bigserial not null, url text not null, hash text not null ); diff --git a/swh/core/db/tests/test_cli.py b/swh/core/db/tests/test_cli.py index a944608..156dabb 100644 --- a/swh/core/db/tests/test_cli.py +++ b/swh/core/db/tests/test_cli.py @@ -1,224 +1,217 @@ # Copyright (C) 2019-2022 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 from datetime import datetime, timezone import pytest from swh.core.cli.db import db as swhdb from swh.core.db import BaseDb -from swh.core.db.pytest_plugin import postgresql_fact from swh.core.tests.test_cli import assert_section_contains def now(): return datetime.now(tz=timezone.utc) def test_cli_swh_help(swhmain, cli_runner): swhmain.add_command(swhdb) result = cli_runner.invoke(swhmain, ["-h"]) assert result.exit_code == 0 assert_section_contains( result.output, "Commands", "db Software Heritage database generic tools." ) help_db_snippets = ( ( "Usage", ( "Usage: swh db [OPTIONS] COMMAND [ARGS]...", "Software Heritage database generic tools.", ), ), ( "Commands", ( "create Create a database for the Software Heritage .", "init Initialize a database for the Software Heritage .", "init-admin Execute superuser-level initialization steps", ), ), ) def test_cli_swh_db_help(swhmain, cli_runner): swhmain.add_command(swhdb) result = cli_runner.invoke(swhmain, ["db", "-h"]) assert result.exit_code == 0 for section, snippets in help_db_snippets: for snippet in snippets: assert_section_contains(result.output, section, snippet) -# We do not want the truncate behavior for those tests -test_db = postgresql_fact( - "postgresql_proc", dbname="clidb", no_truncate_tables={"dbversion", "origin"} -) - - @pytest.fixture -def swh_db_cli(cli_runner, monkeypatch, test_db): +def swh_db_cli(cli_runner, monkeypatch, postgresql): """This initializes a cli_runner and sets the correct environment variable expected by the cli to run appropriately (when not specifying the --dbname flag) """ - db_params = test_db.get_dsn_parameters() + db_params = postgresql.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): +def test_cli_swh_db_create_and_init_db(cli_runner, postgresql, mock_package_sql): """Create a db then initializing it should be ok """ module_name = "test.cli" - conninfo = craft_conninfo(test_db, "new-db") + conninfo = craft_conninfo(postgresql, "new-db") # This creates the db and installs the necessary admin extensions result = cli_runner.invoke(swhdb, ["create", module_name, "--dbname", 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, "--dbname", 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 + cli_runner, postgresql, mock_package_sql ): """Init command on an inexisting db cannot work """ module_name = "test.cli" # it's mocked here - conninfo = craft_conninfo(test_db, "inexisting-db") + conninfo = craft_conninfo(postgresql, "inexisting-db") result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", 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 + cli_runner, postgresql, mock_package_sql ): """Init command cannot work without privileged extension. In this test, the schema needs privileged extension to work. """ module_name = "test.cli" # it's mocked here - conninfo = craft_conninfo(test_db) + conninfo = craft_conninfo(postgresql) result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", 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 + cli_runner, postgresql, mock_package_sql ): """Init commands with carefully crafted libpq conninfo works """ module_name = "test.cli" # it's mocked here - conninfo = craft_conninfo(test_db) + conninfo = craft_conninfo(postgresql) result = cli_runner.invoke(swhdb, ["init-admin", module_name, "--dbname", conninfo]) assert result.exit_code == 0, f"Unexpected output: {result.output}" result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", 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: + with BaseDb.connect(postgresql.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): +def test_cli_swh_db_initialization_with_env(swh_db_cli, mock_package_sql, postgresql): """Init commands with standard environment variables works """ module_name = "test.cli" # it's mocked here cli_runner, db_params = swh_db_cli result = cli_runner.invoke( swhdb, ["init-admin", module_name, "--dbname", db_params["dbname"]] ) assert result.exit_code == 0, f"Unexpected output: {result.output}" result = cli_runner.invoke( swhdb, ["init", module_name, "--dbname", 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: + with BaseDb.connect(postgresql.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): +def test_cli_swh_db_initialization_idempotent(swh_db_cli, mock_package_sql, postgresql): """Multiple runs of the init commands are idempotent """ module_name = "test.cli" # mocked cli_runner, db_params = swh_db_cli result = cli_runner.invoke( swhdb, ["init-admin", module_name, "--dbname", db_params["dbname"]] ) assert result.exit_code == 0, f"Unexpected output: {result.output}" result = cli_runner.invoke( swhdb, ["init", module_name, "--dbname", db_params["dbname"]] ) assert result.exit_code == 0, f"Unexpected output: {result.output}" result = cli_runner.invoke( swhdb, ["init-admin", module_name, "--dbname", db_params["dbname"]] ) assert result.exit_code == 0, f"Unexpected output: {result.output}" result = cli_runner.invoke( swhdb, ["init", module_name, "--dbname", 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: + with BaseDb.connect(postgresql.dsn).cursor() as cur: cur.execute("select * from origin") origins = cur.fetchall() assert len(origins) == 1