tests: Simplify database pytest fixture

Stop using the DatabaseJanitor from pytest-postgresql to start a
separate PostgreSQL server for the tests, but instead connect to
existing database server via the configured database settings and
appending "_test" to the database name.  If there is no configuration
(nor DATABASE_URL env var), fallback to connecting to passari_test
database on local unix socket.

Use functions from sqlalchemy-utils to drop the test database if it
already exists and to (re-)create it to ensure that a fresh test
database is available for each test case.

This makes the database fixture much simpler and allows running the
tests on host that doesn't have PostgreSQL installed (by connecting to
remote PostgreSQL instance).

Author: suutari-ai

requirements_dev.txt

@@ -4,5 +4,5 @@ fakeredis~=2.26.1
 freezegun~=1.5.1
 pytest-asyncio~=0.24.0
 pytest-cov~=6.0.0
-pytest-postgresql~=6.1.1
 lupa~=2.2
+sqlalchemy-utils~=0.41

tests/conftest.py

@@ -1,22 +1,22 @@
 import datetime
-import os
+import re
 import subprocess
 import time
 from pathlib import Path
 
 from sqlalchemy import create_engine
+from sqlalchemy_utils import create_database, database_exists, drop_database
 
 import fakeredis
 import freezegun
 import pytest
 from passari.config import CONFIG as PAS_CONFIG
 from passari_workflow.config import CONFIG
 from passari_workflow.db import DBSession
-from passari_workflow.db.connection import connect_db
+from passari_workflow.db.connection import connect_db, get_connection_uri
 from passari_workflow.db.models import (Base, MuseumAttachment,
                                                MuseumObject, MuseumPackage,
                                                SyncStatus)
-from pytest_postgresql.janitor import DatabaseJanitor
 
 
 @pytest.fixture(scope="function", autouse=True)
@@ -42,68 +42,20 @@ def redis(monkeypatch):
 
 @pytest.fixture(scope="session")
 def database(request):
-    def get_psql_version():
-        result = subprocess.check_output(["psql", "--version"]).decode("utf-8")
-        version = result.split(" ")[-1].strip()
-        major, minor, *_ = version.split(".")
-
-        # Get the major and minor version, which are what pytest-postgresql
-        # wants
-        return f"{major}.{minor}"
-
-    if os.environ.get("POSTGRES_USER"):
-        # Use separately launched process if environments variables are defined
-        # This is used in Gitlab CI tests which run in a Docker container
-        user = os.environ["POSTGRES_USER"]
-        host = os.environ["POSTGRES_HOST"]
-        password = os.environ["POSTGRES_PASSWORD"]
-
-        # POSTGRES_PORT can also be a value such as "tcp://1.1.1.1:5432"
-        # This handles that format as well
-        port = int(os.environ.get("POSTGRES_PORT", "5432").split(":")[-1])
-        db_name = "passari_test"
-        version = os.environ["POSTGRES_VERSION"]
-        create_engine(
-            f"postgresql://{user}:{password}@{host}:{port}/{db_name}"
-        )
-
-        yield request.getfixturevalue("postgresql_nooproc")
-    else:
-        # Launch PostgreSQL ourselves
-        postgresql = request.getfixturevalue("postgresql_proc")
-
-        user = postgresql.user
-        host = postgresql.host
-        port = postgresql.port
-        db_name = "passari_test"
-
-        version = get_psql_version()
-
-        with DatabaseJanitor(
-            user=user,
-            host=host,
-            port=port,
-            dbname=db_name,
-            version=version
-        ):
-            create_engine(
-                f"postgresql://{user}@{host}:{port}/{db_name}"
-            )
-            yield postgresql
+    db_url = get_connection_uri(default="postgresql:///passari")
+    test_db_url = re.sub(r"^([^?]*[^/?])(/?[?]?.*)$", r"\1_test\2", db_url)
+    assert "_test" in test_db_url
+    if database_exists(test_db_url):
+        drop_database(test_db_url)
+    create_database(test_db_url)
+    yield create_engine(test_db_url)
 
 
 @pytest.fixture(scope="function")
 def engine(database, monkeypatch):
-    monkeypatch.setitem(CONFIG["db"], "url", "")
-    monkeypatch.setitem(CONFIG["db"], "user", database.user)
-    monkeypatch.setitem(
-        CONFIG["db"], "password",
-        # Password authentication is used when running tests under Docker
-        os.environ.get("POSTGRES_PASSWORD", "")
-    )
-    monkeypatch.setitem(CONFIG["db"], "host", database.host)
-    monkeypatch.setitem(CONFIG["db"], "port", database.port)
-    monkeypatch.setitem(CONFIG["db"], "name", "passari_test")
+    monkeypatch.setitem(CONFIG["db"], "url", str(database.url))
+    for item in ("user", "password", "host", "port", "name"):
+        monkeypatch.setitem(CONFIG["db"], item, "")
 
     engine = connect_db()
     engine.echo = True