Dans Pytest, le framework de test Python préféré de tous, un appareil est un morceau de code réutilisable qui organise quelque chose avant l'entrée du test et nettoie après sa fermeture. Par exemple, un fichier ou un dossier temporaire, un environnement de configuration, le démarrage d'un serveur Web, etc. Dans cet article, nous verrons comment créer un appareil Pytest qui crée une base de données de test (vide ou avec un état connu) qui obtient nettoyé, permettant à chaque test de s'exécuter sur une base de données complètement propre.

Les objectifs

Nous allons créer un appareil Pytest en utilisant Psycopg 3 pour préparer et nettoyer la base de données de test. Parce qu'une base de données vide est rarement utile pour les tests, nous appliquerons éventuellement les migrations Yoyo (au moment de la rédaction de cet article, le site Web est en panne, accédez à l'instantané archive.org) pour la remplir.

Ainsi, les exigences pour le luminaire Pytest nommé test_db créé dans cet article de blog sont :

• abandonner la base de données de test si elle existe avant le test
• créer une base de données vide avant le test
• éventuellement appliquer des migrations ou créer des données de test avant le test
• fournir une connexion à la base de données de test au test
• abandonner la base de données de test après le test (même en cas d'échec)

Toute méthode de test qui la demande en lui listant un argument de méthode de test :

def test_create_admin_table(test_db):
    ...

Recevra une instance de connexion Psycopg régulière connectée à la base de données de test. Le test peut faire tout ce dont il a besoin, comme avec l'usage courant de Psycopg, par exemple : 

def test_create_admin_table(test_db):
    # Open a cursor to perform database operations
    cur = test_db.cursor()

    # Pass data to fill a query placeholders and let Psycopg perform
    # the correct conversion (no SQL injections!)
    cur.execute(
        "INSERT INTO test (num, data) VALUES (%s, %s)",
        (100, "abc'def"))

    # Query the database and obtain data as Python objects.
    cur.execute("SELECT * FROM test")
    cur.fetchone()
    # will return (1, 100, "abc'def")

    # You can use `cur.fetchmany()`, `cur.fetchall()` to return a list
    # of several records, or even iterate on the cursor
    for record in cur:
        print(record)

Disposition du fichier de projet

Dans un projet Python classique, les sources résident dans src/ et les tests dans tests/:

├── src
│   └── tuvok
│       ├── __init__.py
│       └── sales
│           └── new_user.py
├── tests
│   ├── conftest.py
│   └── sales
│       └── test_new_user.py
├── requirements.txt
└── yoyo.ini

Si vous utilisez une bibliothèque de migrations comme le fantastique Yoyo, les scripts de migration se trouvent probablement dans migrations/:

├── migrations
    ├── 20240816_01_Yn3Ca-sales-user-user-add-last-run-table.py
    ├── ...

Configuration

Notre appareil de base de données de test nécessitera très peu de configuration :

• URL de connexion - (sans base de données)
• nom de la base de données de test - sera recréé pour chaque test
• (facultatif) dossier migrations - scripts de migration à appliquer pour chaque test

Pytest a un emplacement naturel conftest.py pour partager des appareils sur plusieurs fichiers. La configuration des appareils y ira également :

# Without DB name!
TEST_DB_URL = "postgresql://localhost"
TEST_DB_NAME = "test_tuvok"
TEST_DB_MIGRATIONS_DIR = str(Path(__file__, "../../migrations").resolve())

Vous pouvez définir ces valeurs à partir de la variable d'environnement ou tout ce qui convient à votre cas.

Créer un appareil test_db

Avec la connaissance de la bibliothèque PostgreSQL et Psycopg, écrivez le luminaire dans conftest.py :

@pytest.fixture
def test_db():
    # autocommit=True start no transaction because CREATE/DROP DATABASE
    # cannot be executed in a transaction block.
    with psycopg.connect(TEST_DB_URL, autocommit=True) as conn:
        cur = conn.cursor()

        # create test DB, drop before
        cur.execute(f'DROP DATABASE IF EXISTS "{TEST_DB_NAME}" WITH (FORCE)')
        cur.execute(f'CREATE DATABASE "{TEST_DB_NAME}"')

        # Return (a new) connection to just created test DB
        # Unfortunately, you cannot directly change the database for an existing Psycopg connection. Once a connection is established to a specific database, it's tied to that database.
        with psycopg.connect(TEST_DB_URL, dbname=TEST_DB_NAME) as conn:
            yield conn

        cur.execute(f'DROP DATABASE IF EXISTS "{TEST_DB_NAME}" WITH (FORCE)')

Créer un appareil de migration

Dans notre cas, nous utilisons les migrations Yoyo. Écrivez les migrations d'application sous un autre appareil appelé yoyo :

@pytest.fixture
def yoyo():
    # Yoyo expect `driver://user:pass@host:port/database_name?param=value`.
    # In passed URL we need to
    url = (
        urlparse(TEST_DB_URL)
        .
        # 1) Change driver (schema part) with `postgresql psycopg` to use
        # psycopg 3 (not 2 which is `postgresql psycopg2`)
        _replace(scheme="postgresql psycopg")
        .
        # 2) Change database to test db (in which migrations will apply)
        _replace(path=TEST_DB_NAME)
        .geturl()
    )

    backend = get_backend(url)
    migrations = read_migrations(TEST_DB_MIGRATIONS_DIR)

    if len(migrations) == 0:
        raise ValueError(f"No Yoyo migrations found in '{TEST_DB_MIGRATIONS_DIR}'")

    with backend.lock():
        backend.apply_migrations(backend.to_apply(migrations))

Si vous souhaitez appliquer des migrations à chaque base de données de test, exigez le luminaire yoyo pour le luminaire test_db :

@pytest.fixture
def test_db(yoyo):
    ...

Pour appliquer la migration à certains tests uniquement, exigez yoyo individuellement :

def test_create_admin_table(test_db, yoyo):
    ...

Conclusion

Construire mon propre appareil pour donner à vos tests une base de données propre a été une expérience enrichissante pour moi, me permettant d'approfondir à la fois Pytest et Postgres.

J'espère que cet article vous a aidé avec votre propre suite de tests de base de données. N'hésitez pas à me laisser votre question dans les commentaires et bon codage !