Merge pull request #9 from andersinno/test-db-improvement

tests: Simplify database pytest fixture

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