Changeset View
Changeset View
Standalone View
Standalone View
swh/core/db/tests/test_cli.py
| Show First 20 Lines • Show All 67 Lines • ▼ Show 20 Lines | def mock_package_sql(mocker, datadir): | ||||
| mock_sql_files = mocker.patch("swh.core.cli.db.get_sql_for_package") | 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) | sql_files = sorted(glob.glob(path.join(datadir, "cli", "*.sql")), key=sortkey) | ||||
| mock_sql_files.return_value = sql_files | mock_sql_files.return_value = sql_files | ||||
| return mock_sql_files | return mock_sql_files | ||||
| # We do not want the truncate behavior for those tests | # We do not want the truncate behavior for those tests | ||||
| test_db = postgresql_fact( | test_db = postgresql_fact( | ||||
| "postgresql_proc", db_name="clidb", no_truncate_tables={"dbversion", "origin"} | "postgresql_proc", dbname="clidb", no_truncate_tables={"dbversion", "origin"} | ||||
| ) | ) | ||||
| @pytest.fixture | @pytest.fixture | ||||
| def swh_db_cli(cli_runner, monkeypatch, test_db): | def swh_db_cli(cli_runner, monkeypatch, test_db): | ||||
| """This initializes a cli_runner and sets the correct environment variable expected by | """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) | the cli to run appropriately (when not specifying the --dbname flag) | ||||
| """ | """ | ||||
| db_params = test_db.get_dsn_parameters() | db_params = test_db.get_dsn_parameters() | ||||
| monkeypatch.setenv("PGHOST", db_params["host"]) | monkeypatch.setenv("PGHOST", db_params["host"]) | ||||
| monkeypatch.setenv("PGUSER", db_params["user"]) | monkeypatch.setenv("PGUSER", db_params["user"]) | ||||
| monkeypatch.setenv("PGPORT", db_params["port"]) | monkeypatch.setenv("PGPORT", db_params["port"]) | ||||
| return cli_runner, db_params | return cli_runner, db_params | ||||
| Show All 14 Lines | |||||
| 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, test_db, mock_package_sql): | ||||
| """Create a db then initializing it should be ok | """Create a db then initializing it should be ok | ||||
| """ | """ | ||||
| module_name = "something" | module_name = "something" | ||||
| conninfo = craft_conninfo(test_db, "new-db") | conninfo = craft_conninfo(test_db, "new-db") | ||||
| # This creates the db and installs the necessary admin extensions | # This creates the db and installs the necessary admin extensions | ||||
| result = cli_runner.invoke(swhdb, ["create", module_name, "--db-name", conninfo]) | result = cli_runner.invoke(swhdb, ["create", module_name, "--dbname", conninfo]) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| # This initializes the schema and data | # This initializes the schema and data | ||||
| result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo]) | result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", conninfo]) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| # the origin value in the scripts uses a hash function (which implementation wise | # 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) | # uses a function from the pgcrypt extension, installed during db creation step) | ||||
| with BaseDb.connect(conninfo).cursor() as cur: | with BaseDb.connect(conninfo).cursor() as cur: | ||||
| cur.execute("select * from origin") | cur.execute("select * from origin") | ||||
| origins = cur.fetchall() | origins = cur.fetchall() | ||||
| assert len(origins) == 1 | assert len(origins) == 1 | ||||
| def test_cli_swh_db_initialization_fail_without_creation_first( | def test_cli_swh_db_initialization_fail_without_creation_first( | ||||
| cli_runner, test_db, mock_package_sql | cli_runner, test_db, mock_package_sql | ||||
| ): | ): | ||||
| """Init command on an inexisting db cannot work | """Init command on an inexisting db cannot work | ||||
| """ | """ | ||||
| module_name = "anything" # it's mocked here | module_name = "anything" # it's mocked here | ||||
| conninfo = craft_conninfo(test_db, "inexisting-db") | conninfo = craft_conninfo(test_db, "inexisting-db") | ||||
| result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo]) | result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", conninfo]) | ||||
| # Fails because we cannot connect to an inexisting db | # Fails because we cannot connect to an inexisting db | ||||
| assert result.exit_code == 1, f"Unexpected output: {result.output}" | assert result.exit_code == 1, f"Unexpected output: {result.output}" | ||||
| def test_cli_swh_db_initialization_fail_without_extension( | def test_cli_swh_db_initialization_fail_without_extension( | ||||
| cli_runner, test_db, mock_package_sql | cli_runner, test_db, mock_package_sql | ||||
| ): | ): | ||||
| """Init command cannot work without privileged extension. | """Init command cannot work without privileged extension. | ||||
| In this test, the schema needs privileged extension to work. | In this test, the schema needs privileged extension to work. | ||||
| """ | """ | ||||
| module_name = "anything" # it's mocked here | module_name = "anything" # it's mocked here | ||||
| conninfo = craft_conninfo(test_db) | conninfo = craft_conninfo(test_db) | ||||
| result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo]) | result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", conninfo]) | ||||
| # Fails as the function `public.digest` is not installed, init-admin calls is needed | # Fails as the function `public.digest` is not installed, init-admin calls is needed | ||||
| # first (the next tests show such behavior) | # first (the next tests show such behavior) | ||||
| assert result.exit_code == 1, f"Unexpected output: {result.output}" | assert result.exit_code == 1, f"Unexpected output: {result.output}" | ||||
| def test_cli_swh_db_initialization_works_with_flags( | def test_cli_swh_db_initialization_works_with_flags( | ||||
| cli_runner, test_db, mock_package_sql | cli_runner, test_db, mock_package_sql | ||||
| ): | ): | ||||
| """Init commands with carefully crafted libpq conninfo works | """Init commands with carefully crafted libpq conninfo works | ||||
| """ | """ | ||||
| module_name = "anything" # it's mocked here | module_name = "anything" # it's mocked here | ||||
| conninfo = craft_conninfo(test_db) | conninfo = craft_conninfo(test_db) | ||||
| result = cli_runner.invoke( | result = cli_runner.invoke(swhdb, ["init-admin", module_name, "--dbname", conninfo]) | ||||
| swhdb, ["init-admin", module_name, "--db-name", conninfo] | |||||
| ) | |||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| result = cli_runner.invoke(swhdb, ["init", module_name, "--db-name", conninfo]) | result = cli_runner.invoke(swhdb, ["init", module_name, "--dbname", conninfo]) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| # the origin values in the scripts uses a hash function (which implementation wise | # 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) | # uses a function from the pgcrypt extension, init-admin calls installs it) | ||||
| with BaseDb.connect(test_db.dsn).cursor() as cur: | with BaseDb.connect(test_db.dsn).cursor() as cur: | ||||
| cur.execute("select * from origin") | cur.execute("select * from origin") | ||||
| origins = cur.fetchall() | origins = cur.fetchall() | ||||
| assert len(origins) == 1 | 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, test_db): | ||||
| """Init commands with standard environment variables works | """Init commands with standard environment variables works | ||||
| """ | """ | ||||
| module_name = "anything" # it's mocked here | module_name = "anything" # it's mocked here | ||||
| cli_runner, db_params = swh_db_cli | cli_runner, db_params = swh_db_cli | ||||
| result = cli_runner.invoke( | result = cli_runner.invoke( | ||||
| swhdb, ["init-admin", module_name, "--db-name", db_params["dbname"]] | swhdb, ["init-admin", module_name, "--dbname", db_params["dbname"]] | ||||
| ) | ) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| result = cli_runner.invoke( | result = cli_runner.invoke( | ||||
| swhdb, ["init", module_name, "--db-name", db_params["dbname"]] | swhdb, ["init", module_name, "--dbname", db_params["dbname"]] | ||||
| ) | ) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| # the origin values in the scripts uses a hash function (which implementation wise | # 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) | # uses a function from the pgcrypt extension, init-admin calls installs it) | ||||
| with BaseDb.connect(test_db.dsn).cursor() as cur: | with BaseDb.connect(test_db.dsn).cursor() as cur: | ||||
| cur.execute("select * from origin") | cur.execute("select * from origin") | ||||
| origins = cur.fetchall() | origins = cur.fetchall() | ||||
| assert len(origins) == 1 | 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, test_db): | ||||
| """Multiple runs of the init commands are idempotent | """Multiple runs of the init commands are idempotent | ||||
| """ | """ | ||||
| module_name = "anything" # mocked | module_name = "anything" # mocked | ||||
| cli_runner, db_params = swh_db_cli | cli_runner, db_params = swh_db_cli | ||||
| result = cli_runner.invoke( | result = cli_runner.invoke( | ||||
| swhdb, ["init-admin", module_name, "--db-name", db_params["dbname"]] | swhdb, ["init-admin", module_name, "--dbname", db_params["dbname"]] | ||||
| ) | ) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| result = cli_runner.invoke( | result = cli_runner.invoke( | ||||
| swhdb, ["init", module_name, "--db-name", db_params["dbname"]] | swhdb, ["init", module_name, "--dbname", db_params["dbname"]] | ||||
| ) | ) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| result = cli_runner.invoke( | result = cli_runner.invoke( | ||||
| swhdb, ["init-admin", module_name, "--db-name", db_params["dbname"]] | swhdb, ["init-admin", module_name, "--dbname", db_params["dbname"]] | ||||
| ) | ) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| result = cli_runner.invoke( | result = cli_runner.invoke( | ||||
| swhdb, ["init", module_name, "--db-name", db_params["dbname"]] | swhdb, ["init", module_name, "--dbname", db_params["dbname"]] | ||||
| ) | ) | ||||
| assert result.exit_code == 0, f"Unexpected output: {result.output}" | assert result.exit_code == 0, f"Unexpected output: {result.output}" | ||||
| # the origin values in the scripts uses a hash function (which implementation wise | # 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) | # uses a function from the pgcrypt extension, init-admin calls installs it) | ||||
| with BaseDb.connect(test_db.dsn).cursor() as cur: | with BaseDb.connect(test_db.dsn).cursor() as cur: | ||||
| cur.execute("select * from origin") | cur.execute("select * from origin") | ||||
| origins = cur.fetchall() | origins = cur.fetchall() | ||||
| assert len(origins) == 1 | assert len(origins) == 1 | ||||