Drop test database from noproc fixture, if exists - closes #265

Author: fizyk

README.rst

@@ -227,6 +227,18 @@ You can pick which you prefer, but remember that these settings are handled in t
      - postgresql_options
      - yes
      -
+   * - Drop test database on start.
+
+       .. warning::
+
+           Use carefully as it might lead to unexpected results within your test suite.
+     -
+     - --postgresql-drop-test-database
+     -
+     - false
+     -
+
+
 
 
 Example usage:

newsfragments/265.feature.1.rst

@@ -0,0 +1,3 @@
+DatabaseJanitor.cursor now accepts optional parameter dbname, which defaults to `postgres`
+
+This database name is used to make connection to and return a cursor.

newsfragments/265.feature.rst

@@ -0,0 +1,3 @@
+If a test run ended in an error that prevented proper test cleanup,
+developer can now use `--postgresql-drop-test-database` command line flag,
+to delete database from noproc fixture at the start.

pytest_postgresql/config.py

@@ -21,6 +21,7 @@ class PostgresqlConfigDict(TypedDict):
     dbname: str
     load: List[Union[Path, str]]
     postgres_options: str
+    drop_test_database: bool
 
 
 def get_config(request: FixtureRequest) -> PostgresqlConfigDict:
@@ -45,6 +46,7 @@ def get_postgresql_option(option: str) -> Any:
         dbname=get_postgresql_option("dbname"),
         load=load_paths,
         postgres_options=get_postgresql_option("postgres_options"),
+        drop_test_database=request.config.getoption("postgresql_drop_test_database"),
     )
 
 

pytest_postgresql/factories/client.py

@@ -23,6 +23,7 @@
 from psycopg import Connection
 from pytest import FixtureRequest
 
+from pytest_postgresql.config import get_config
 from pytest_postgresql.executor import PostgreSQLExecutor
 from pytest_postgresql.executor_noop import NoopExecutor
 from pytest_postgresql.janitor import DatabaseJanitor
@@ -52,14 +53,15 @@ def postgresql_factory(request: FixtureRequest) -> Iterator[Connection]:
         proc_fixture: Union[PostgreSQLExecutor, NoopExecutor] = request.getfixturevalue(
             process_fixture_name
         )
+        config = get_config(request)
 
         pg_host = proc_fixture.host
         pg_port = proc_fixture.port
         pg_user = proc_fixture.user
         pg_password = proc_fixture.password
         pg_options = proc_fixture.options
         pg_db = dbname or proc_fixture.dbname
-        with DatabaseJanitor(
+        janitor = DatabaseJanitor(
             user=pg_user,
             host=pg_host,
             port=pg_port,
@@ -68,7 +70,10 @@ def postgresql_factory(request: FixtureRequest) -> Iterator[Connection]:
             version=proc_fixture.version,
             password=pg_password,
             isolation_level=isolation_level,
-        ):
+        )
+        if config["drop_test_database"]:
+            janitor.drop()
+        with janitor:
             db_connection: Connection = psycopg.connect(
                 dbname=pg_db,
                 user=pg_user,

pytest_postgresql/factories/noprocess.py

@@ -72,6 +72,7 @@ def postgresql_noproc_fixture(request: FixtureRequest) -> Iterator[NoopExecutor]
         pg_dbname = xdistify_dbname(dbname or config["dbname"])
         pg_options = options or config["options"]
         pg_load = load or config["load"]
+        drop_test_database = config["drop_test_database"]
 
         noop_exec = NoopExecutor(
             host=pg_host,
@@ -81,14 +82,17 @@ def postgresql_noproc_fixture(request: FixtureRequest) -> Iterator[NoopExecutor]
             dbname=pg_dbname,
             options=pg_options,
         )
-        with DatabaseJanitor(
+        janitor = DatabaseJanitor(
             user=noop_exec.user,
             host=noop_exec.host,
             port=noop_exec.port,
             template_dbname=noop_exec.template_dbname,
             version=noop_exec.version,
             password=noop_exec.password,
-        ) as janitor:
+        )
+        if drop_test_database is True:
+            janitor.drop()
+        with janitor:
             for load_element in pg_load:
                 janitor.load(load_element)
             yield noop_exec

pytest_postgresql/janitor.py

@@ -127,12 +127,12 @@ def load(self, load: Union[Callable, str, Path]) -> None:
         )
 
     @contextmanager
-    def cursor(self) -> Iterator[Cursor]:
+    def cursor(self, dbname: str = "postgres") -> Iterator[Cursor]:
         """Return postgresql cursor."""
 
         def connect() -> Connection:
             return psycopg.connect(
-                dbname="postgres",
+                dbname=dbname,
                 user=self.user,
                 password=self.password,
                 host=self.host,

pytest_postgresql/plugin.py

@@ -34,6 +34,11 @@
 _help_dbname = "Default database name"
 _help_load = "Dotted-style or entrypoint-style path to callable or path to SQL File"
 _help_postgres_options = "Postgres executable extra parameters. Passed via the -o option to pg_ctl"
+_help_drop_test_database = (
+    "Drop test database in noproc and client fixture, for the cases, "
+    "when database was not cleared due to errors in previous test runs. "
+    "Use cautiously and not on CI."
+)
 
 
 def pytest_addoption(parser: Parser) -> None:
@@ -127,6 +132,13 @@ def pytest_addoption(parser: Parser) -> None:
         help=_help_postgres_options,
     )
 
+    parser.addoption(
+        "--postgresql-drop-test-database",
+        action="store_true",
+        dest="postgresql_drop_test_database",
+        help=_help_drop_test_database,
+    )
+
 
 postgresql_proc = factories.postgresql_proc()
 postgresql_noproc = factories.postgresql_noproc()

tests/examples/test_drop_test_database.py

@@ -0,0 +1,16 @@
+"""All tests for pytest-postgresql."""
+
+from psycopg import Connection
+
+from pytest_postgresql import factories
+
+postgresql = factories.postgresql("postgresql_noproc")
+
+
+def test_postgres_load_override(postgresql: Connection) -> None:
+    """Check postgresql fixture can load one file and override database 'leftover'."""
+    cur = postgresql.cursor()
+    cur.execute("SELECT * FROM test;")
+    results = cur.fetchall()
+    assert len(results) == 1
+    cur.close()

tests/test_postgres_options_plugin.py

@@ -6,6 +6,11 @@
 from pytest import Pytester
 
 import pytest_postgresql
+from pytest_postgresql.executor import PostgreSQLExecutor
+from pytest_postgresql.factories import postgresql_proc
+from pytest_postgresql.factories.noprocess import xdistify_dbname
+from pytest_postgresql.janitor import DatabaseJanitor
+from tests.loader import load_database
 
 
 @pytest.fixture
@@ -41,3 +46,78 @@ def test_postgres_loader_in_cli(pointed_pytester: Pytester) -> None:
     test_sql_path = pointed_pytester.copy_example("test.sql")
     ret = pointed_pytester.runpytest(f"--postgresql-load={test_sql_path}", "test_load.py")
     ret.assert_outcomes(passed=1)
+
+
+postgresql_proc_to_override = postgresql_proc()
+
+
+def test_postgres_drop_test_database(
+    postgresql_proc_to_override: PostgreSQLExecutor,
+    pointed_pytester: Pytester,
+) -> None:
+    """Check that the database is dropped on both process and client level if argument is passed.
+
+    Given:
+        Preexisting tables override_tmpl and override (created with two DatabaseJanitor instances)
+    When:
+        Run test that connects to the process from current process with flag to drop database
+        specified
+    Then:
+        The internal pytest run will delete the database on start (and after).
+        It Would fail if it did not. Checks are performed by trying to connect to both override_tmpl
+        and override databases and checking resulting exceptions.
+    """
+    dbname = xdistify_dbname("override")
+    template_dbname = dbname + "_tmpl"
+    template_janitor = DatabaseJanitor(
+        user=postgresql_proc_to_override.user,
+        host=postgresql_proc_to_override.host,
+        port=postgresql_proc_to_override.port,
+        template_dbname=template_dbname,
+        version=postgresql_proc_to_override.version,
+        password=postgresql_proc_to_override.password,
+        connection_timeout=5,
+    )
+    template_janitor.init()
+    template_janitor.load(load_database)
+    assert template_janitor.template_dbname
+    janitor = DatabaseJanitor(
+        user=postgresql_proc_to_override.user,
+        host=postgresql_proc_to_override.host,
+        port=postgresql_proc_to_override.port,
+        dbname=dbname,
+        template_dbname=template_janitor.template_dbname,
+        version=postgresql_proc_to_override.version,
+        password=postgresql_proc_to_override.password,
+        connection_timeout=5,
+    )
+    janitor.init()
+    assert janitor.dbname
+    with janitor.cursor(janitor.dbname) as cur:
+        cur.execute("SELECT * FROM stories")
+        res = cur.fetchall()
+        assert len(res) == 4
+    # Actual test happens now
+    pointed_pytester.copy_example("test_drop_test_database.py")
+    test_sql_path = pointed_pytester.copy_example("test.sql")
+    ret = pointed_pytester.runpytest(
+        f"--postgresql-load={test_sql_path}",
+        f"--postgresql-port={postgresql_proc_to_override.port}",
+        "--postgresql-dbname=override",
+        "--postgresql-drop-test-database",
+        "test_drop_test_database.py",
+    )
+    ret.assert_outcomes(passed=1)
+
+    with pytest.raises(TimeoutError) as excinfo:
+        with janitor.cursor(janitor.dbname):
+            pass
+    assert hasattr(excinfo.value, "__cause__")
+    assert f'FATAL:  database "{janitor.dbname}" does not exist' in str(excinfo.value.__cause__)
+    with pytest.raises(TimeoutError) as excinfo:
+        with template_janitor.cursor(template_janitor.template_dbname):
+            pass
+    assert hasattr(excinfo.value, "__cause__")
+    assert f'FATAL:  database "{template_janitor.template_dbname}" does not exist' in str(
+        excinfo.value.__cause__
+    )