Sylvain Lequeux

Passionné par la qualité du code dans le monde du Data Engineering et du Machine Learning.

Pratiques et conseils pour gagner en sérénité en partant d'une base de code data non testée.

Nous allons commencer cet article par une expérience de pensée. Imaginons un projet data qui réunirait un grand nombre de mauvaises pratiques du monde du développement. Nous y trouverons des fonctions de plusieurs centaines de lignes, qui mélangent lecture et écriture de la donnée au traitement business. Pas de chance, ce projet possède également des notebooks. Dans ces notebooks, certaines cellules lisent de la donnée, mais les chemins des fichiers à charger font référence au disque local de Paul, Data Scientist dans l’équipe. Ils ne sont donc pas exploitables par les autres membres de l’équipe.

Cette situation à première vue digne d’un sénario de Cauchemard en cuisine est malheureusement plutôt fréquent. Je suis intervenu par le passé sur plusieurs projets qui se sentaient dans une impasse suite à une base de code assez proche de ce que j’ai décrit. Mon rôle était alors de leur permettre de sortir la tête de l’eau, de gagner en sérénité quand les membres de l’équipe touchaient au code. Nous allons ici présenter un certain nombre de pratiques et de conseils pour aborder cette problématique.

Commencer par un test global

Améliorer la qualité de cette base de code va passer par une étape dans laquelle nous modifierons le code sans y ajouter de fonctionnalité. Cependant, si l’équipe n’a pas la confiance nécessaire sur sa base de code, alors dérouler cette étape peut s’avérer dangereux.

Pour se rassurer, il faudra donc commencer par écrire des tests avant même de changer le code. Mais là aussi, on se retrouve souvent en difficulité. Que tester quand les fonctions font plusieurs centaines de lignes ? Quand elles sont fortement dépendantes les unes des autres ?

Rendre la base de code testable

Il est grand temps de passer à la pratique. Prenons cet exemple de code Python qui va lire un fichier CSV, calculer un aggrégat et le stocker dans un autre fichier.

import pandas as pd

if __name__ == 'main':
    df = pd.read_csv('/home/slequeux/my_dataset/some_file.csv')
    df_agg = df # Remplacer cette ligne par un nombre important de transformations métier
    df_agg.to_csv('/home/slequeux/my_dataset/file_with_agg.csv')

Nous pouvons d’ores et déjà remarquer plusieurs éléments problématiques sur ce code :

• Ce code ne dispose d’aucune fonction que l’on pourrait tester.
• Ce code lit des fichiers dont le chemin est spécifié en dur.
• Ce code mélange des lectures / écritures disque au code métier.

La première étape, afin de pouvoir écrire notre premier test, sera de déplacer ces quelques lignes de code dans une fonction à l’aide d’un refactoring automatisé par l’IDE, le seul type de refactoring autorisé sans test (car sans risque d’erreur) :

import pandas as pd

def main() -> None:
    df = pd.read_csv('/home/slequeux/my_dataset/some_file.csv')
    df_agg = df # Remplacer cette ligne par un nombre important de transformations métier
    df_agg.to_csv('/home/slequeux/my_dataset/file_with_agg.csv')

if __name__ == 'main':
    main()

Nous allons maintenant pouvoir écrire un test global. Ce test va chercher à valider au mieux le comportement général de notre programme. Il ne validera malheureusement pas tous les cas possibles, mais au moins quelques cas nominaux.

Les différentes étapes d’un test peuvent être décrites en utilisant le découpage Given / When / Then (notamment utilisé dans la syntaxe Gherkin).

La partie When est la plus simple à remplir, il suffira d’appeler la fonction main.

from main_package import main

def test_main_function():
    # Given
    # décrit le contexte d'exécution du test

    # When
    main()

    # Then
    # décrit des assertions sur le résultat recherché

Fournir la donnée en entrée

Pour la partie Given, on se retrouve à devoir posséder un fichier en local dans un répertoire bien spécifique. Cela étant un peu dérangeant, nous allons extraire les chemins écrits en durs en tant que paramètres de la fonction main à l’aide d’un refactoring automatisé par l’IDE, le seul type de refactoring autorisé sans test (car sans risque d’erreur).

import pandas as pd

def main(input_file: str = '/home/slequeux/my_dataset/some_file.csv', 
         output_file: str = '/home/slequeux/my_dataset/file_with_agg.csv') -> None:
    df = pd.read_csv(input_file)
    df_agg = df # Remplacer cette ligne par un nombre important de transformations métier
    df_agg.to_csv(output_file)

if __name__ == 'main':
    main()

Il est maintenant possible de créer un fichier d’exemple dans un répertoire de test que l’on fournira à cette fonction main. On peut par exemple utiliser pytest dans le monde de Python, ou bien utiliser une bibliothèque équivalente, ou encore coder ce comportement à la main dans le test. Prenons un fichier CSV d’exemple, plaçons le dans le répertoire local tests/examples.

from pathlib import Path

from main_package import main

def test_main_function(tmp_path: Path):
    # Given some input file
    example_file = Path('.') / 'tests/examples/some_file.csv'

    # And a target file
    output_path = tmp_path / 'output_file.csv'
    
    # When
    main(example_file, output_path)

    # Then
    # décrit des assertions sur le résultat recherché

Conseil : minimiser la taille du jeu de données d’entrée

Essayez de minimiser la taille du jeu de données d’entrée. Plus l’entrée sera grosse, plus il sera compliqué de la maintenir. On préférera écrire plusieurs petits datasets d’inputs et dupliquer le test pour une meilleure converture plutôt que de mettre tous les subsets dans le même jeu de données d’entrée.

Suivant la complexité du code, réduire la taille du jeu de données d’entrée peut être assez compliqué. Si vraiment vous n’y arrivez pas, le jeu de données « réel » peut alors être fourni si vous n’êtes pas dans un contexte Big Data.

Valider la sortie

Reste maintenant à écrire les assertions.

Ecrire manuellement le jeu de données attendu en sortie du test sera probablement trop complexe. Nous allons pour cela nous baser sur une approche décrite dans l’article Testing legacy code with Golden Master. Etant donné que la sortie de notre test est déjà un fichier et non pas un objet, nous allons juste adapter un peu cette approche.

Jouez une première fois le test, mais en spécifiant un fichier output_path qui n’est pas temporaire pour qu’il ne soit pas effacé à la fin de l’exécution dudit test. Récupérez ce fichier et placez le dans le répertoire tests/examples à côté du fichier d’input. Il est maintenant possible de finaliser l’écriture du test.

from pathlib import Path

import pandas as pd
from main_package import main

def test_main_function(tmp_path: Path):
    # Given some input file
    example_file = Path('.') / 'test/examples/some_file.csv'

    # And a target file
    output_path = tmp_path / 'output_file.csv'
    
    # When
    main(example_file, output_path)

    # Then
    produced_df = pd.read_csv(output_path)
    expected_df = pd.read_csv(Path('.') / 'test/examples/recorded_output.csv')
    pd.testing.assert_frame_equal(expected_df, produced_df)

Vous avez ainsi un premier test global. Si, dans le futur, le refactoring modifie le comportement business, le fichier généré va changer et le test échouera. Vous êtes donc un minimum couverts pour toucher à la base de code.

Bon, dans le vrai monde réel de la réalité véritable, il reste encore quelques détails :

• On a testé ici un cas de la base de code principale. Il y aura probablement des cas particuliers, des branchements, etc. Il conviendra d’écrire quelques tests pour se rassurer. Mais on ne pourra pas ici obtenir une couverture de 100% sans dépenser un temps dont on ne dispose généralement pas. Nous allons donc faire un compromis entre risque et temps passé.
• Vous aurez probablement besoins de plusieurs fichiers en entrée. Il suffira alors de faire plusieurs zones Given qui vont s’enchaîner.

Cas particulier d’accès à une base de données

Pour la création de ce test, il nous reste à voir le cas particulier d’un accès à une base de données par votre fonction main.

Dans le cas idéal, il serait préférable d’écrire un premier test global en utilisant les méthodes décrites précédemment et en utillisant potentiellement votre base de données d’un environnement de développement. Une fois ce test écrit, ou si vous n’êtes pas en mesure de l’écrire (impossibilité d’accès à une base de données depuis votre poste ou depuis la machine de build par exemple), la procédure consistera à s’abstraire d’une connexion à une vraie base de données, évidemment en touchant le moins possible à votre code.

Imaginons cette partie de la fonction principale :

def main() -> None:
    # Some code here
    my_db_driver = MyDBDriver.initializeDriver()
    my_db_driver.write(some_database_record)
    # Some code here

if __name__ == 'main':
    main()

La première étape consistera à sortir l’instanciation du driver de base de données de la fonction.

def main(db_driver: MyDBDriver) -> None:
    # Some code here
    db_driver.write(some_database_record)
    # Some code here

if __name__ == 'main':
    my_db_driver = MyDBDriver.initializeDriver()
    main(my_db_driver)

Une fois cela fait, vous pouvez créer un wrapper autour du driver qui va s’occuper de faire un passe plat.

class MySuperDbDriverWrapper:
    def __init__(self, db_driver: MyDBDriver):
        self.db_driver = db_driver

    def write(self, record) -> None:
        self.db_driver.write(record)

def main(db_driver: MySuperDbDriverWrapper) -> None:
    # Some code here
    db_driver.write(some_database_record)
    # Some code here

if __name__ == 'main':
    driver_wrapper = MySuperDbDriverWrapper(MyDBDriver.initializeDriver())
    main(driver_wrapper)

Puis extraire une interface de cette nouvelle classe.

from abc import ABC, abstractmethod

class DbDriverWrapper(ABC):

    @abstractmethod
    def write(self, record) -> None:
        pass

class MySuperDbDriverWrapper(DbDriverWrapper):
    def __init__(self, db_driver):
        self.db_driver = db_driver

    def write(self, record) -> None:
        self.db_driver.write(record)

def main(db_driver: DbDriverWrapper) -> None:
    # Some code here
    db_driver.write(some_database_record)
    # Some code here

if __name__ == 'main':
    driver_wrapper = MySuperDbDriverWrapper(MyDBDriver.initializeDriver())
    main(driver_wrapper)

Dans votre test, il sera maintenant possible de fournir en paramètre du main une instanciation locale qui enregistre les appels à la méthode pour les valider par la suite

class SpyDbDriver(DbDriverWrapper):
    def __init__(self):
            self.writes = []

    def write(self, record):
        self.writes.append(record)

def test_main_function():
    # Given
    db_driver = SpyDbDriver()

    # When
    main(db_driver)

    # Then
    assert(len(db_driver.writes) == 1)

Suivant le besoin à observer dans votre implémentation de fausse base de données, il conviendra d’adapter la classe SpyDbDriver.

Vient l’étape de refactoring

Vous disposez maintenant d’un (ou de quelques) test(s) de haut niveau qui vont vous permettre de toucher à la base de code avec un peu plus de sérénité. Vient donc maintenant l’étape de refactoring.

Cette étape itérative est composée de trois parties :

1. Choisir une fonctionnalité de votre base de code ;
2. Extraire cette fonctionnalité dans une plus petite fonction ;
3. Tester unitairement la fonction.

Il peut encore arriver que la fonctionnalité ainsi extraite soit encore trop grosse pour être testée. Vous pourrez alors appliquer la méthode vue en début d’article sur celle-ci pour la tester à haut niveau avant de la refactorer.

Une montagne de travail à gravir caillou par caillou

Tester unitairement l’intégralité de sa base de code en partant d’une couverture à inexistante représente une montagne de travail. Voici donc deux astuces complémentaires pour s’engager sur cette voie que vous pourrez mettre en place après avoir écrits quelques tests de bout en bout comme précisé précédemment dans cet article.

Technique de la bulle d’air

Identifiez une petite partie du code, une petite fonctionnalité. Testez complètement cette petite partie. Vous pourrez maintenant considérer cette partie comme une bulle d’air pour les développeurs dans laquelle ils travailleront de manière sereine.

Lorsque vous voulez tester une nouvelle partie de votre code, tentez de tester une zone proche de cette partie du code. Cela vous permettra à la fois de réduire la complexité de l’écriture de ces nouveaux tests (tout ce qui touchera au code déjà refactoré devrait être plutôt facilement testable), mais aussi d’augmenter la taille de la bulle d’air pour la suite du projet.

La règle du boy scout

La règle du boy scout est une pratique qui consiste à toujours laisser la base de code dans un état meilleur qu’à son arrivée.

• Si vous ajoutez une fonctionnalité, alors cette fonctionnalité doit être testée ;
• Si vous touchez à une partie du code existant, alors ajoutez des tests à cette partie.

Les deux techniques que nous avons vu pourront vous aider dans votre voyage vers cette terre promise d’un code propre et testable. Ne cherchez pas immédiatement une couverture à 100 %. Si vous le faites, alors vous ne ferez que tester. Vous allez alors y passer beaucoup de temps, temps pendant lequel vous ne pourrez plus ajouter de nouvelles fonctionnalités ni corriger d’éventuels bugs. Le produit que vous construisez doit continuer de vivre tout en ayant une qualité qui s’améliore avec le temps.

La montagne de travail évoquée plus tôt doit donc être vue comme une succession de petits cailloux à gravir.

Concernant les notebooks

Les pratiques que nous avons vu jusqu’ici peuvent être appliquées à presque n’importe quel domaine du développement logiciel.

Nous allons maintenant nous attarder sur un élément bien spécifique au monde de la Data : les notebooks. A grands coups de chemins écrits en dur, de cellules qui ne sont pas exécutées dans l’ordre, les notebooks ont une facheuse tendance à rapidement contenir du code qui n’est pas reproductible.

Pour être concerné par ce que nous allons voir ici, je considérerai que vos notebooks sont correctement versionnés dans Git.

Si vous souhaitez maintenant que tout se passe pour le mieux, il faudra réunir trois conditions :

• Le code à l’intérieur des notebooks doit utiliser au maximum les fonctions de votre base de code principale, qui est (ou sera dans un avenir proche si vous suivez les conseils précédents) testée
• Vous utilisez un IDE suffisament intelligent pour vous accompagner dans le refactoring.
• Vos notebooks ne sont pas stockés sous un format *.ipynb mais sous un format de script Python.

Ainsi, si vous renommez une fonction, supprimez un paramètre, extrayez une classe ou tout autre action habituelle du refactoring, alors votre IDE s’occupera pour vous de mettre à jour le notebook !

Vous avez maintenant les armes pour affronter un code legacy

Améliorer petit à petit la qualité d’une base de code legacy permettra rarement de voir émerger un design minimaliste comme le ferait une approche TDD, mais parfois nous n’avons pas le choix.

Avec toutes ces techniques, vous devriez avoir les armes pour affronter un code data legacy, et ne plus être en panique devant la complexité qu’il représente.

• Commencez par un test global de votre fonction principale. Ce test doit simuler tout accès à un système externe comme des fichiers ou une base de données ;
• Refactorez petit à petit sans Big Bang ;
• Utilisez Jupytext pour sauvegarder vos notebooks en tant que script Python ;
• Utilisez votre base de code testée dans vos notebooks.