Comment tester les dépendances externes avec Pytest, Docker et Tox | de Vargas Herrera Daniel | janvier 2023

Posté Par: Windows2022tutosur: janvier 02, 2023Dans: Node JsPas De Commentaires

Écrivez des tests qui s’appuient sur des dépendances conteneurisées et orchestrez-les avec Tox

photo par ThisisEngineering RAEng au Unsplash

Lors du développement d’une nouvelle application, nous avons souvent besoin de services externes tels que des bases de données, des courtiers de messages, de la mémoire cache ou des API vers d’autres micro-services. Tester les interfaces entre ces composants est une pratique souvent négligée. Certains sautent complètement ces tests, et d’autres se moquent des réponses. Cependant, cela laisse les tests d’intégration incomplets, car ces interfaces peuvent facilement échouer en raison de modifications des API des fournisseurs, de modifications du schéma de base de données et de problèmes de configuration.

Une façon d’effectuer ces tests consiste à exécuter les dépendances en tant que conteneurs qui communiquent avec l’application dans un environnement similaire à la production. Dans cet article, j’illustre comment exécuter ces tests dans un pipeline CI à l’aide d’actions GitHub. Pour ce faire, vous devez écrire une petite application Python avec accès à une base de données postgres pour créer et sélectionner des utilisateurs.

Les tests seront écrits en utilisant pytest, l’un des principaux frameworks de test pour Python. La base de données sera exécutée en tant que conteneur à l’aide de Docker. Les tests seront coordonnés avec toxiqueun orchestrateur de tests pour Python, qui est apparu à l’origine comme un outil pour tester différentes versions de Python, et a évolué pour permettre aux développeurs d’isoler les tests des environnements de développement, de centraliser la configuration des tests et même de coordonner les dépendances à l’aide de conteneurs Docker.

Vous pouvez suivre les instructions et les extraits de code ou consulter l’ensemble (mais petit) du projet directement sur mon Dépôt GitHub. La structure de fichier du projet est celle-ci :

.
├── README.md
├── migrations
│   └── schema.sql
├── requirements.txt
├── src
│   └── testing_containers
│       ├── __init__.py
│       ├── db
│       │   ├── __init__.py
│       │   └── db.py
│       └── model
│           ├── __init__.py
│           └── users.py
├── tests
│   ├── __init__.py
│   ├── conftest.py
│   └── db
│       └── test_db.py
└── tox.ini

Puisque nous utilisons une base de données Postgres comme dépendance, commençons par écrire un petit tableau représentant Users dans notre application. je vais l’appeler migrations/schema.sql. Voici à quoi ça ressemble :

CREATE TABLE users (
email varchar(64) primary key,
name varchar(64) not null
);

Pour représenter les entités dans l’application, j’utiliserai pydantic . C’est un excellent moyen de représenter des objets, d’obtenir des indications de type, et il est particulièrement utile d’effectuer des validations de données lorsque nous utilisons ces entités dans des API. Alors, créez un nouveau fichier Python, appelez-le users.pyet collez ce qui suit :

from pydantic import BaseModelclass User(BaseModel):
"""Representation of User entity"""
name: str
email: str

Nous pouvons maintenant écrire un passe-partout pour nos opérations de base de données. Commençons par quelques fonctions qui enveloppent notre pilote postgres ( psycopg2 dans ce cas). Comme l’intention de cet article n’est pas d’enseigner comment utiliser ce pilote, je ne m’étendrai pas dessus, mais je vous encouragerai à lire le Tutoriel PYnatif. Nous pouvons commencer notre cours comme ceci :

from typing import Listimport psycopg2
from psycopg2.extras import execute_values
from testing_containers.model.users import User
class Repo:
# psycopg2 wrapper
def __init__(self) -> None:
self.conn = None
try:
self.connect()
logging.info("db: database ready")
except Exception as err:
logging.error("db: failed to connect to database: %s", str(err))
self.close()
raise err
def connect(self):
"""Stores a connection object `conn` of a postgres database"""
logging.info("db: connecting to database")
conn_str = os.environ.get("DB_DSN")
self.conn = psycopg2.connect(conn_str)
def close(self):
"""Closes the connection object `conn`"""
if self.conn is not None:
logging.info("db: closing database")
self.conn.close()
def execute_select_query(self, query: str, args: tuple = ()) -> List[tuple]:
"""Executes a read query and returns the result
Args:
query (str): the query to execute.
args (tuple, optional): arguments to the select statement. Defaults to ().
Returns:
List[tuple]: result of the select statement. One element per record
"""
with self.conn.cursor() as cur:
cur.execute(query, args)
return list(cur)
def execute_multiple_insert_query(self, query: str, data: List[tuple], page_size: int = 100) -> None:
"""Execute a statement using :query:`VALUES` with a sequence of parameters.
Args:
query (str): the query to execute. It must contain a single ``%s``
placeholder, which will be replaced by a `VALUES list`__.
Example: ``"INSERT INTO table (id, f1, f2) VALUES %s"``.
data (List[tuple]): sequence of sequences or dictionaries with the arguments to send to the query.
page_size (int, optional): maximum number of *data* items to include in every statement.
If there are more items the function will execute more than one statement. Defaults to 100.
"""
with self.conn.cursor() as cur:
execute_values(cur, query, data, page_size=page_size)

Maintenant, nous écrivons quelques fonctions pour :

Récupérer un utilisateur par nom
Créer de nouveaux utilisateurs

def get_user(self, name: str) -> (User | None):
"""Retrieve a User by a name"""
query = "SELECT name, email FROM users WHERE name = %s"
res = self.execute_select_query(query, (name,))
if len(res) == 0:
return None
record = res[0]
return User(name=record[0], email=record[1])def insert_users(self, users: List[User]) -> None:
"""Given a list of Users, insert them in the db"""
query = "INSERT INTO users (name, email) VALUES %s"
data: List[tuple] = [(u.name, u.email) for u in users]
self.execute_multiple_insert_query(query, data)

Impressionnant! Nous avons certaines fonctions qui interagissent sans entités commerciales.

Notez que si nous ne testons pas ces fonctions, nous sommes susceptibles d’erreurs si le schéma change ou si nous violons les contraintes de la base de données.

Alors, commençons les tests !

import pytest
from psycopg2.errors import UniqueViolation@pytest.mark.usefixtures("repo")
class TestRepo:
def test_insert_users(self):
repo: Repo = self.repo # repository instanced passed by the fixture
# define some users
alice = User(name="alice", email="alice@example.com")
bob = User(name="bob", email="bob@example.com")
robert = User(name="robert", email="bob@example.com")
# check that the users are not there at first
result = repo.get_user("alice")
assert result is None
# check that the users are there after inserting
users = [alice, bob]
repo.insert_users(users)
result = repo.get_user("alice")
assert result == alice
result = repo.get_user("bob")
assert result == bob
# check that pk fails
with pytest.raises(UniqueViolation):
repo.insert_users([robert])

La première ligne est un décorateur qui spécifie que notre classe de test TestRepo utilisera un appareil appelé repo. Agencements sont des fonctions de base qui nous permettent de produire des résultats de test cohérents et reproductibles.

Comprenons cette ligne repo: Repo = self.repo. Nous créons une variable appelée repo de type Repo qui est notre classe de référentiel précédemment définie. Il est réaffecté de self.repoqui vient de l’appareil, comme je l’expliquerai plus tard.

Nous affirmons deux choses :

Il n’y a pas d’utilisateurs au début du test
Alice et Bob les utilisateurs sont récupérés après l’insertion
Il y a un UniqueViolation exception, et elle est déclenchée lorsque nous tentons de créer un nouvel utilisateur avec un e-mail déjà existant.

Écrivons notre appareil dans un nouveau conftest.py dossier.

@pytest.fixture(scope="class", name="repo")
def repo(request):
"""Instantiates a database object"""
db = Repo()
try:
request.cls.repo = db
yield db
finally:
db.close()

Dans cet appareil, nous procédons comme suit :

Instancier notre classe de référentiel db = Repo .
Dans un try block (car l’exécution du test peut déclencher une exception), nous utilisons le pytest’s request fixture pour que notre classe de test puisse accéder au référentiel request.cls.repo = db . Je vous encourage à en savoir plus sur le request fixation!
Ensuite, nous donnons l’instance à utiliser dans les fonctions de test.
Dans un finally block, nous nous assurons de fermer la connexion à la base de données.

Super! Nous avons nos tests, mais si vous êtes impatient de courir pytest déjà, vous verrez que les tests échouent au niveau du fixture, car nous ne pouvons pas instancier le dépôt sans une base de données postgres !

C’est ici que la toxicomanie vient à la rescousse. Créez un nouveau fichier appelé tox.ini:

[tox]
envlist = py310[testenv]
setenv =
DB_DSN = postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable
deps =
-r requirements.txt
commands = pytest

Ce fichier de configuration minimal nous indique que nous effectuerons des tests avec un environnement Python 3.10. Il définit une variable environnementale DB_DSN, spécifie un fichier d’exigences à installer dans un environnement virtuel et appelle pytest. Mon fichier d’exigences ressemble à ceci:

psycopg2-binary
pydantic
pytest

Au fait, je recommande d’utiliser pip-toos pour épingler les dépendances. Cela sort du cadre de ce tutoriel, mais vous pouvez le lire ici : .

Exécuter vos tests maintenant est aussi simple que d’installer et d’exécuter tox:

python -m pip install — user tox

tox

Bien sûr, cela échoue car j’ai promis que tox coordonnerait un conteneur postgres, et je ne l’ai pas fait.

Tox est un outil utile qui peut devenir puissant avec ses plugins. Tox docker est l’un d’entre eux, et il s’installe facilement en exécutant la commande suivante :

pip install tox-docker

Maintenant, nous pouvons étendre notre tox.ini. Voici comment procéder :

[tox]
envlist = py310[testenv]
setenv =
DB_DSN = postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable
deps =
-r requirements.txt
commands = pytest
docker = postgres
[docker:postgres]
image = postgres:13.4-alpine
environment =
POSTGRES_DB=postgres
PGUSER=postgres
POSTGRES_PASSWORD=postgres
POSTGRES_HOST_AUTH_METHOD=trust
ports =
5432:5432/tcp
healthcheck_cmd = pg_isready
healthcheck_timeout = 5
healthcheck_retries = 5
healthcheck_interval = 5
healthcheck_start_period = 1
volumes =
bind:ro:{toxinidir}/migrations/schema.sql:/docker-entrypoint-initdb.d/0.init.sql

Remarquez que dans notre testenv section, nous spécifions que nous utiliserons un conteneur docker nommé postgres que nous définissons immédiatement après. Nous définissons l’image docker qu’il doit (extraire) utiliser, les variables environnementales, les ports à mapper, les vérifications de l’état (utiles pour s’assurer que nos tests ne s’exécutent que lorsque nos conteneurs sont sains) et les volumes (notez que je me réfère àmigrations/schema.sql qui contient notre définition de table SQL). Vérifiez s’il vous plaît documentation tox-docker si vous voulez plus de détails.

Maintenant, en courant tox nous réussissons nos tests !

Notez que tox crée les conteneurs, exécute les tests, puis supprime les conteneurs pour nous. Plutôt cool, non ?

Exécuter nos tests localement est une excellente pratique, mais pour être très minutieux, il est préférable de les exécuter également dans notre outil de contrôle de version lors d’une demande d’extraction. Avec Actions GitHub, nous pouvons créer un petit pipeline CI pour exécuter tox chaque fois qu’un commit est poussé vers une demande d’extraction. Créez simplement ce fichier, /.github/worflows/pr-test.yaml :

name: PR Teston:
pull_request:
branches:
- main
jobs:   
test:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.10"]
name: With Python ${{ matrix.python-version }}
steps:
- name: Checkout code
uses: actions/checkout@v2
- name: Setup python
uses: actions/setup-python@v2
with:
python-version: ${{ matrix.python-version }}
architecture: x64
- name: Install dependencies
run: |
python -m pip install --upgrade pip
python -m pip install tox tox-gh-actions
- name: Test with tox
run: tox

Désormais, chaque fois que nous créons une demande d’extraction dans notre base de code, GitHub Actions s’exécute tox et s’assure que nos tests réussissent.

Avec l’approche actuelle, nous pouvons facilement faire évoluer les tests de dépendance en ajoutant plus de conteneurs docker dans la configuration tox.

[docker:redis]
image = bitnami/redis:latest
environment =
ALLOW_EMPTY_PASSWORD=yes
REDIS_PORT_NUMBER=7000
ports =
7000:7000/tcp
healthcheck_cmd = redis-cli ping
healthcheck_timeout = 5
healthcheck_retries = 5
healthcheck_interval = 5
healthcheck_start_period = 1

Les dépendances externes qui ne peuvent pas être conteneurisées ou qui représentent des coûts ou des ressources protégées, comme une API privée, devraient mieux être simulées.
Si l’utilité ne vous convainc pas de tox, une approche différente consiste à définir les dépendances dans un fichier docker-compose, créer et exécuter les services, attendre qu’ils soient sains, exécuter les pytests et arrêter et supprimer gracieusement les conteneurs.