Changeset View
Changeset View
Standalone View
Standalone View
swh/core/db/tests/pytest_plugin/test_pytest_plugin.py
| # Copyright (C) 2020 The Software Heritage developers | # Copyright (C) 2020 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 | ||||
| import glob | |||||
| import os | import os | ||||
| from swh.core.db import BaseDb | from swh.core.db import BaseDb | ||||
| from swh.core.db.pytest_plugin import postgresql_fact | from swh.core.db.pytest_plugin import postgresql_fact | ||||
| SQL_DIR = os.path.join(os.path.dirname(__file__), "data") | SQL_DIR = os.path.join(os.path.dirname(__file__), "data") | ||||
| # db with special policy for tables dbversion and people | # db with special policy for tables dbversion and people | ||||
| postgres_fun = postgresql_fact( | postgres_fun = postgresql_fact( | ||||
| "postgresql_proc", | "postgresql_proc", | ||||
| db_name="fun", | db_name="fun", | ||||
| dump_files=f"{SQL_DIR}/*.sql", | dump_files=f"{SQL_DIR}/*.sql", | ||||
| no_truncate_tables={"dbversion", "people"}, | no_truncate_tables={"dbversion", "people"}, | ||||
| ) | ) | ||||
| postgres_fun2 = postgresql_fact( | |||||
| "postgresql_proc", | |||||
vlorentz: fun fact (pun intended): if you write `db_name="fun"` instead of `db_name="fun2"`, it makes… | |||||
| db_name="fun2", | |||||
| dump_files=sorted(glob.glob(f"{SQL_DIR}/*.sql")), | |||||
| no_truncate_tables={"dbversion", "people"}, | |||||
| ) | |||||
| def test_smoke_test_fun_db_is_up(postgres_fun): | def test_smoke_test_fun_db_is_up(postgres_fun): | ||||
| """This ensures the db is created and configured according to its dumps files. | """This ensures the db is created and configured according to its dumps files. | ||||
| """ | """ | ||||
| with BaseDb.connect(postgres_fun.dsn).cursor() as cur: | with BaseDb.connect(postgres_fun.dsn).cursor() as cur: | ||||
| cur.execute("select count(*) from dbversion") | cur.execute("select count(*) from dbversion") | ||||
| nb_rows = cur.fetchone()[0] | nb_rows = cur.fetchone()[0] | ||||
| assert nb_rows == 5 | assert nb_rows == 5 | ||||
| cur.execute("select count(*) from fun") | cur.execute("select count(*) from fun") | ||||
| nb_rows = cur.fetchone()[0] | nb_rows = cur.fetchone()[0] | ||||
| assert nb_rows == 3 | assert nb_rows == 3 | ||||
| cur.execute("select count(*) from people") | cur.execute("select count(*) from people") | ||||
| nb_rows = cur.fetchone()[0] | nb_rows = cur.fetchone()[0] | ||||
| assert nb_rows == 2 | assert nb_rows == 2 | ||||
| # in data, we requested a value already so it starts at 2 | # in data, we requested a value already so it starts at 2 | ||||
| cur.execute("select nextval('serial')") | cur.execute("select nextval('serial')") | ||||
| val = cur.fetchone()[0] | val = cur.fetchone()[0] | ||||
| assert val == 2 | assert val == 2 | ||||
| def test_smoke_test_fun2_db_is_up(postgres_fun2): | |||||
| """This ensures the db is created and configured according to its dumps files. | |||||
| """ | |||||
| with BaseDb.connect(postgres_fun2.dsn).cursor() as cur: | |||||
| cur.execute("select count(*) from dbversion") | |||||
| nb_rows = cur.fetchone()[0] | |||||
| assert nb_rows == 5 | |||||
| cur.execute("select count(*) from fun") | |||||
| nb_rows = cur.fetchone()[0] | |||||
| assert nb_rows == 3 | |||||
| cur.execute("select count(*) from people") | |||||
| nb_rows = cur.fetchone()[0] | |||||
| assert nb_rows == 2 | |||||
| # in data, we requested a value already so it starts at 2 | |||||
| cur.execute("select nextval('serial')") | |||||
| val = cur.fetchone()[0] | |||||
| assert val == 2 | |||||
| def test_smoke_test_fun_db_is_still_up_and_got_reset(postgres_fun): | def test_smoke_test_fun_db_is_still_up_and_got_reset(postgres_fun): | ||||
| """This ensures that within another tests, the 'fun' db is still up, created (and not | """This ensures that within another tests, the 'fun' db is still up, created (and not | ||||
| configured again). This time, most of the data has been reset: | configured again). This time, most of the data has been reset: | ||||
| - except for tables 'dbversion' and 'people' which were left as is | - except for tables 'dbversion' and 'people' which were left as is | ||||
| - the other tables from the schema (here only "fun") got truncated | - the other tables from the schema (here only "fun") got truncated | ||||
| - the sequences got truncated as well | - the sequences got truncated as well | ||||
| """ | """ | ||||
| ▲ Show 20 Lines • Show All 89 Lines • Show Last 20 Lines | |||||
fun fact (pun intended): if you write db_name="fun" instead of db_name="fun2", it makes pytest freeze and ignore SIGTERM.