python Twisted의 adbapi가 unittests에서 데이터를 복구하지 못하는 이유는 무엇입니까?



unit-testing sqlite3 (2)

좋아, 이건 좀 까다 롭다. 테스트를 분리하여 실행하면 (이 질문에 게시 된 것처럼) 버그가 거의 발생하지 않게됩니다. 그러나 전체 테스트 제품군의 컨텍스트에서 실행하면 거의 100 % 실패합니다.

DB에 쓰고 DB에서 읽기 전에 yield task.deferLater(reactor, .00001, lambda: None) 를 추가했습니다. 그러면이 문제가 해결됩니다.

거기에서, 나는 이것이 커넥션 풀과 sqlite의 제한된 동시성 - 허용 오차 (concurrency-tolerance)에 기인 한 경쟁 조건일지도 모른다고 생각했다. ConnectionPoolcb_mincb_max 매개 변수를 1 로 설정하려고 시도했지만 문제도 해결되었습니다.

간단히 말해서 : sqlite가 여러 연결에서 매우 잘 실행되지 않는 것처럼 보이며 적절한 수정은 가능한 범위까지 동시성을 피하는 것입니다.

개요

문맥

나는 SQLite3 데이터베이스에 쓰는 것에 의존하는 고차 논리에 대한 단위 테스트를 작성하고있다. 이를 위해 나는 twisted.trial.unittesttwisted.enterprise.adbapi.ConnectionPool 사용하고 있습니다.

문제 설명

영구 sqlite3 데이터베이스를 만들고 거기에 데이터를 저장할 수 있습니다. sqlitebrowser를 사용하여 데이터가 예상대로 유지되었는지 확인할 수 있습니다.

문제는 teaConnectionPool.run* (예 : runQuery )에 대한 호출은 TestCase 내에서 호출 된 경우에만 빈 결과 집합을 반환한다는 것입니다.

메모 및 중요 정보

내가 겪고있는 문제는 Twisted의 trial 프레임 워크에서만 발생합니다. 디버깅을 시도한 첫 번째 시도는 데이터베이스 코드를 유닛 테스트에서 꺼내어 독립적 인 테스트 / 디버그 스크립트에 배치하는 것이 었습니다. 이 스크립트는 단위 테스트 코드가 작동하지 않는 동안 예상대로 작동합니다 (아래 예제 참조).

사례 1 : 부적절한 단위 테스트

init.sql

이것은 데이터베이스를 초기화하는 데 사용되는 스크립트입니다. 이 파일에서 발생하는 (명백한) 오류는 없습니다.

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

예기치 않게 실패한 단위 테스트 클래스입니다. TestStateManagement.test_db_clean 이 전달되어 테이블이 제대로 작성되었음을 나타냅니다. TestStateManagement.test_inode_create 가 실패하여 0 개의 결과가 검색되었다는 것을 나타냅니다.

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

이들은 위의 단위 테스트에 의해 테스트되는 인공물입니다.

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)

사례 2 : 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")

맺음말

다시 sqlitebrowser로 확인하면 데이터가 db.sqlite 에 쓰여지는 것처럼 보이므로 검색 문제처럼 보입니다. 여기부터, 나는 어떤 종류의 저주 받았어 ... 어떤 생각?

편집하다

이 코드는 테스트에 사용할 수있는 inode 를 생성합니다.

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",
}

setUp 함수를 self.db.runInteraction(...) 을 반환하고 지연을 반환합니다. 앞에서 언급했듯이 지연이 완료 될 때까지 기다리는 것으로 가정합니다. 그러나 이것은 사실이 아니며 대부분의 가을 희생자가 포함 된 함정입니다 (자신 포함). 솔직히 말해서, 이런 상황, 특히 단위 테스트의 경우, 나는 단지 TestCase 클래스 외부의 동기 코드를 실행하여 데이터베이스를 초기화합니다. 예 :

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
    """

또는 설치 프로그램을 yield runOperation(...) 할 수 있지만 작동하지 않는다고 알려주는 경우가 있습니다 ... 어떤 경우에도 오류가 발생하지 않은 것은 놀라운 일입니다.

추신

나는 잠시 동안이 질문에 눈을 떴다. 그리고 그것은 나의 머리의 뒤에서 일 동안 지금 있었다. 이것에 대한 잠재적 인 이유가 마침내 거의 오전 1시에 시작되었습니다. 그러나, 나는 이것을 실제로 시험하기에는 너무 지치거나 게으르다 : D하지만 그것은 꽤 지긋 지긋하다. 이 질문에서 당신의 수준에 대해 당신에게 추천합니다.





python-db-api