unitaire - unittest python 3




Pourquoi l'adbapi de Twisted ne parvient-il pas à récupérer des données à l'intérieur de tests unitaires? (2)

D'accord, il s'avère que c'est un peu délicat. Exécuter les tests de manière isolée (comme cela a été posté à cette question) fait en sorte que le bogue se produit rarement. Cependant, lors de l'exécution dans le contexte d'une suite de tests complète, il échoue presque 100% du temps.

J'ai ajouté yield task.deferLater(reactor, .00001, lambda: None) après avoir écrit sur la base de données et avant de lire la base de données, et cela résout le problème.

De là, j'ai soupçonné que cela pourrait être une condition de concurrence découlant du pool de connexion et de la tolérance de concurrence limitée de sqlite. J'ai essayé de définir les paramètres cb_min et cb_max à ConnectionPool sur 1 , et cela a également résolu le problème.

En bref: il semble que sqlite ne joue pas très bien avec plusieurs connexions, et que le correctif approprié est d'éviter la concurrence dans la mesure du possible.

Aperçu

Le contexte

J'écris des tests unitaires pour une logique d'ordre supérieur qui dépend de l'écriture dans une base de données SQLite3. Pour cela j'utilise twisted.trial.unittest et twisted.enterprise.adbapi.ConnectionPool .

Déclaration de problème

Je suis capable de créer une base de données sqlite3 persistante et y stocker des données. En utilisant sqlitebrowser , je suis capable de vérifier que les données ont été persistées comme prévu.

Le problème est que les appels à teaConnectionPool.run* (par exemple: runQuery ) renvoient un ensemble de résultats vide, mais seulement lorsqu'ils sont appelés depuis un TestCase .

Notes et détails importants

Le problème que je rencontre se produit uniquement dans le cadre d' trial de Twisted. Ma première tentative de débogage consistait à retirer le code de la base de données du test unitaire et à le placer dans un script de test / débogage indépendant. Ce script fonctionne comme prévu alors que le code de test unitaire ne fonctionne pas (voir les exemples ci-dessous).

Cas 1: test unitaire mal conduit

init.sql

C'est le script utilisé pour initialiser la base de données. Il n'y a pas d'erreurs (apparentes) provenant de ce fichier.

CREATE TABLE ajxp_changes ( seq INTEGER PRIMARY KEY AUTOINCREMENT, node_id NUMERIC, type TEXT, source TEXT, target TEXT, deleted_md5 TEXT );
CREATE TABLE ajxp_index ( node_id INTEGER PRIMARY KEY AUTOINCREMENT, node_path TEXT, bytesize NUMERIC, md5 TEXT, mtime NUMERIC, stat_result BLOB);
CREATE TABLE ajxp_last_buffer ( id INTEGER PRIMARY KEY AUTOINCREMENT, type TEXT, location TEXT, source TEXT, target TEXT );
CREATE TABLE ajxp_node_status ("node_id" INTEGER PRIMARY KEY  NOT NULL , "status" TEXT NOT NULL  DEFAULT 'NEW', "detail" TEXT);
CREATE TABLE events (id INTEGER PRIMARY KEY AUTOINCREMENT, type text, message text, source text, target text, action text, status text, date text);

CREATE TRIGGER LOG_DELETE AFTER DELETE ON ajxp_index BEGIN INSERT INTO ajxp_changes (node_id,source,target,type,deleted_md5) VALUES (old.node_id, old.node_path, "NULL", "delete", old.md5); END;
CREATE TRIGGER LOG_INSERT AFTER INSERT ON ajxp_index BEGIN INSERT INTO ajxp_changes (node_id,source,target,type) VALUES (new.node_id, "NULL", new.node_path, "create"); END;
CREATE TRIGGER LOG_UPDATE_CONTENT AFTER UPDATE ON "ajxp_index" FOR EACH ROW BEGIN INSERT INTO "ajxp_changes" (node_id,source,target,type) VALUES (new.node_id, old.node_path, new.node_path, CASE WHEN old.node_path = new.node_path THEN "content" ELSE "path" END);END;
CREATE TRIGGER STATUS_DELETE AFTER DELETE ON "ajxp_index" BEGIN DELETE FROM ajxp_node_status WHERE node_id=old.node_id; END;
CREATE TRIGGER STATUS_INSERT AFTER INSERT ON "ajxp_index" BEGIN INSERT INTO ajxp_node_status (node_id) VALUES (new.node_id); END;

CREATE INDEX changes_node_id ON ajxp_changes( node_id );
CREATE INDEX changes_type ON ajxp_changes( type );
CREATE INDEX changes_node_source ON ajxp_changes( source );
CREATE INDEX index_node_id ON ajxp_index( node_id );
CREATE INDEX index_node_path ON ajxp_index( node_path );
CREATE INDEX index_bytesize ON ajxp_index( bytesize );
CREATE INDEX index_md5 ON ajxp_index( md5 );
CREATE INDEX node_status_status ON ajxp_node_status( status );

test_sqlite.py

C'est la classe de test unitaire qui échoue de façon inattendue. TestStateManagement.test_db_clean passes, a indiqué que les tables ont été correctement créées. TestStateManagement.test_inode_create échoue, indiquant que les résultats zéro ont été récupérés.

import os.path as osp

from twisted.internet import defer
from twisted.enterprise import adbapi

import sqlengine # see below

class TestStateManagement(TestCase):

    def setUp(self):
        self.meta = mkdtemp()

        self.db = adbapi.ConnectionPool(
            "sqlite3", osp.join(self.meta, "db.sqlite"), check_same_thread=False,
        )
        self.stateman = sqlengine.StateManager(self.db)

        with open("init.sql") as f:
            script = f.read()

        self.d = self.db.runInteraction(lambda c, s: c.executescript(s), script)

    def tearDown(self):
        self.db.close()
        del self.db
        del self.stateman
        del self.d

        rmtree(self.meta)

    @defer.inlineCallbacks
    def test_db_clean(self):
        """Canary test to ensure that the db is initialized in a blank state"""

        yield self.d  # wait for db to be initialized

        q = "SELECT name FROM sqlite_master WHERE type='table' AND name=?;"
        for table in ("ajxp_index", "ajxp_changes"):
            res = yield self.db.runQuery(q, (table,))
            self.assertTrue(
                len(res) == 1,
                "table {0} does not exist".format(table)
         )

    @defer.inlineCallbacks
    def test_inode_create_file(self):
        yield self.d

        path = osp.join(self.ws, "test.txt")
        with open(path, "wt") as f:
            pass

        inode = mk_dummy_inode(path)
        yield self.stateman.create(inode, directory=False)

        entry = yield self.db.runQuery("SELECT * FROM ajxp_index")
        emsg = "got {0} results, expected 1.  Are canary tests failing?"
        lentry = len(entry)
        self.assertTrue(lentry == 1, emsg.format(lentry))

sqlengine.py

Ce sont les artefacts testés par les tests unitaires ci-dessus.

def values_as_tuple(d, *param):
    """Return the values for each key in `param` as a tuple"""
    return tuple(map(d.get, param))


class StateManager:
    """Manages the SQLite database's state, ensuring that it reflects the state
    of the filesystem.
    """

    log = Logger()

    def __init__(self, db):
        self._db = db

    def create(self, inode, directory=False):
        params = values_as_tuple(
            inode, "node_path", "bytesize", "md5", "mtime", "stat_result"
        )

        directive = (
            "INSERT INTO ajxp_index (node_path,bytesize,md5,mtime,stat_result) "
            "VALUES (?,?,?,?,?);"
        )

        return self._db.runOperation(directive, params)

Cas 2: le bug disparaît en dehors de twisted.trial

#! /usr/bin/env python

import os.path as osp
from tempfile import mkdtemp

from twisted.enterprise import adbapi
from twisted.internet.task import react
from twisted.internet.defer import inlineCallbacks

INIT_FILE = "example.sql"


def values_as_tuple(d, *param):
    """Return the values for each key in `param` as a tuple"""
    return tuple(map(d.get, param))


def create(db, inode):
    params = values_as_tuple(
        inode, "node_path", "bytesize", "md5", "mtime", "stat_result"
    )

    directive = (
        "INSERT INTO ajxp_index (node_path,bytesize,md5,mtime,stat_result) "
        "VALUES (?,?,?,?,?);"
    )

    return db.runOperation(directive, params)


def init_database(db):
    with open(INIT_FILE) as f:
        script = f.read()

    return db.runInteraction(lambda c, s: c.executescript(s), script)


@react
@inlineCallbacks
def main(reactor):
    meta = mkdtemp()
    db = adbapi.ConnectionPool(
        "sqlite3", osp.join(meta, "db.sqlite"), check_same_thread=False,
    )

    yield init_database(db)

    # Let's make sure the tables were created as expected and that we're
    # starting from a blank slate
    res = yield db.runQuery("SELECT * FROM ajxp_index LIMIT 1")
    assert not res, "database is not empty [ajxp_index]"

    res = yield db.runQuery("SELECT * FROM ajxp_changes LIMIT 1")
    assert not res, "database is not empty [ajxp_changes]"

    # The details of this are not important.  Suffice to say they (should)
    # conform to the DB schema for ajxp_index.
    test_data = {
        "node_path": "/this/is/some/arbitrary/path.ext",
        "bytesize": 0,
        "mtime": 179273.0,
        "stat_result": b"this simulates a blob of raw binary data",
        "md5": "d41d8cd98f00b204e9800998ecf8427e",  # arbitrary
    }

    # store the test data in the ajxp_index table
    yield create(db, test_data)

    # test if the entry exists in the db
    entry = yield db.runQuery("SELECT * FROM ajxp_index")
    assert len(entry) == 1, "got {0} results, expected 1".format(len(entry))

    print("OK")

Remarques de clôture

Encore une fois, lors de la vérification avec sqlitebrowser, il semble que les données sont écrites dans db.sqlite , donc cela ressemble à un problème de récupération . D'ici, je suis en quelque sorte perplexe ... des idées?

MODIFIER

Ce code produira un inode qui pourra être utilisé pour les tests.

def mk_dummy_inode(path, isdir=False):
return {
    "node_path": path,
    "bytesize": osp.getsize(path),
    "mtime": osp.getmtime(path),
    "stat_result": dumps(stat(path), protocol=4),
    "md5": "directory" if isdir else "d41d8cd98f00b204e9800998ecf8427e",
}

Si vous jetez un oeil à votre fonction setUp , vous renvoyez self.db.runInteraction(...) , qui renvoie un différé. Comme vous l'avez noté, vous supposez qu'il attend que le différé se termine. Cependant, ce n'est pas le cas et c'est un piège dont la plupart sont victimes (y compris moi-même). Je vais être honnête avec vous, pour des situations comme celle-ci, en particulier pour les tests unitaires, j'exécute simplement le code synchrone en dehors de la classe TestCase pour initialiser la base de données. Par exemple:

def init_db():
    import sqlite3
    conn = sqlite3.connect('db.sqlite')
    c = conn.cursor()
    with open("init.sql") as f:
        c.executescript(f.read())

init_db()     # call outside test case


class TestStateManagement(TestCase):
    """
    My test cases
    """

Alternativement, vous pouvez décorer la configuration et yield runOperation(...) mais quelque chose me dit que ça ne marcherait pas ... En tout cas, il est surprenant qu'aucune erreur n'ait été soulevée.

PS

J'ai regardé cette question depuis un moment et ça fait des jours que je suis dans la tête. Une raison potentielle pour cela est finalement apparue sur moi à presque 1h du matin. Cependant, je suis trop fatigué / paresseux pour tester cela: D mais c'est une très bonne idée. Je voudrais vous féliciter pour votre niveau de détail dans cette question.





python-db-api