模拟sqlobject函数调用测试db

Question

我试图用pytest模拟sqlbuilder.func的测试用例。

我成功地用正确的输出模拟了sqlbuilder.func.TO_BASE64，但是当我尝试模拟sqlbuilder.func.FROM_UNIXTIME时，我没有得到任何错误，但是结果的输出与生成的查询不正确。下面是这个问题的最小工作示例。

models.py

from sqlobject import (
    sqlbuilder,
    sqlhub,
    SQLObject,
    StringCol,
    BLOBCol,
    TimestampCol,
)

class Store(SQLObject):
    name = StringCol()
    sample = BLOBCol()
    createdAt = TimestampCol()

DATE_FORMAT = "%Y-%m-%d"
def retrieve(name):
    query = sqlbuilder.Select([
            sqlbuilder.func.TO_BASE64(Store.q.sample),
        ],
        sqlbuilder.AND(
            Store.q.name == name,
            sqlbuilder.func.FROM_UNIXTIME(Store.q.createdAt, DATE_FORMAT) >= sqlbuilder.func.FROM_UNIXTIME("2018-10-12", DATE_FORMAT)
        )
    )

    connection = sqlhub.getConnection()
    query = connection.sqlrepr(query)
    print(query)
    queryResult = connection.queryAll(query)
    return queryResult

conftest.py

import pytest

from models import Store
from sqlobject import sqlhub
from sqlobject.sqlite import sqliteconnection

@pytest.fixture(autouse=True, scope="session")
def sqlite_db_session(tmpdir_factory):
    file = tmpdir_factory.mktemp("db").join("sqlite.db")
    conn = sqliteconnection.SQLiteConnection(str(file))
    sqlhub.processConnection = conn
    init_tables()
    yield conn
    conn.close()

def init_tables():
    Store.createTable(ifNotExists=True)

test_ex1.py

import pytest

from sqlobject import sqlbuilder
from models import retrieve

try:
    import mock
    from mock import MagicMock
except ImportError:
    from unittest import mock
    from unittest.mock import MagicMock

def TO_BASE64(x):
    return x

def FROM_UNIXTIME(x, y):
    return 'strftime("%Y%m%d", datetime({},"unixepoch", "localtime"))'.format(x)

# @mock.patch("sqlobject.sqlbuilder.func.TO_BASE64")
# @mock.patch("sqlobject.sqlbuilder.func.TO_BASE64", MagicMock(side_effect=lambda x: x))
# @mock.patch("sqlobject.sqlbuilder.func.TO_BASE64", new_callable=MagicMock(side_effect=lambda x: x))
@mock.patch("sqlobject.sqlbuilder.func.TO_BASE64", TO_BASE64)
@mock.patch("sqlobject.sqlbuilder.func.FROM_UNIXTIME", FROM_UNIXTIME)
def test_retrieve():
    result = retrieve('Some')
    assert result == []

当前SQL:

SELECT store.sample FROM store WHERE (((store.name) = ('Some')) AND (1))

预期的SQL:

SELECT
  store.sample
FROM 
  store
WHERE
  store.name = 'Some'
AND
  strftime(
    '%Y%m%d',
    datetime(store.created_at, 'unixepoch', 'localtime')
  ) >= strftime(
    '%Y%m%d',
    datetime('2018-10-12', 'unixepoch', 'localtime')
  )

编辑示例

#! /usr/bin/env python

from sqlobject import *

__connection__ = "sqlite:/:memory:?debug=1&debugOutput=1"

try:
    import mock
    from mock import MagicMock
except ImportError:
    from unittest import mock
    from unittest.mock import MagicMock

class Store(SQLObject):
    name = StringCol()
    sample = BLOBCol()
    createdAt = TimestampCol()

Store.createTable()

DATE_FORMAT = "%Y-%m-%d"
def retrieve(name):
    query = sqlbuilder.Select([
            sqlbuilder.func.TO_BASE64(Store.q.sample),
        ],
        sqlbuilder.AND(
            Store.q.name == name,
            sqlbuilder.func.FROM_UNIXTIME(Store.q.createdAt, DATE_FORMAT) >= sqlbuilder.func.FROM_UNIXTIME("2018-10-12", DATE_FORMAT)
        )
    )

    connection = Store._connection
    query = connection.sqlrepr(query)
    queryResult = connection.queryAll(query)
    return queryResult


def TO_BASE64(x):
    return x

def FROM_UNIXTIME(x, y):
    return 'strftime("%Y%m%d", datetime({},"unixepoch", "localtime"))'.format(x)

for p in [
    mock.patch("sqlobject.sqlbuilder.func.TO_BASE64",TO_BASE64),
    mock.patch("sqlobject.sqlbuilder.func.FROM_UNIXTIME",FROM_UNIXTIME),
]:
    p.start()

retrieve('Some')

mock.patch.stopall()

Answer 1

默认情况下，sqlbuilder.func是传递其属性的SQLExpression (例如，sqlbuilder.func.datetime)作为常量(sqlbuilder.func实际上是sqlbuilder.ConstantSpace的别名)。请参阅关于SQLExpression、常见问题和func代码的文档。

当您在func命名空间中模拟一个属性时，它由SQLObject计算并以简化的形式传递给后端。如果您想从模拟函数返回字符串文字，您需要告诉SQLObject它是一个必须按原样传递给后端的值。这样做的方法是用SQLConstant包装文字，如下所示：

def FROM_UNIXTIME(x, y):
    return sqlbuilder.SQLConstant('strftime("%Y%m%d", datetime({},"unixepoch", "localtime"))'.format(x))

见SQLConstant。

整个测试脚本现在看起来如下

#! /usr/bin/env python3.7

from sqlobject import *

__connection__ = "sqlite:/:memory:?debug=1&debugOutput=1"

try:
    import mock
    from mock import MagicMock
except ImportError:
    from unittest import mock
    from unittest.mock import MagicMock

class Store(SQLObject):
    name = StringCol()
    sample = BLOBCol()
    createdAt = TimestampCol()

Store.createTable()

DATE_FORMAT = "%Y-%m-%d"
def retrieve(name):
    query = sqlbuilder.Select([
            sqlbuilder.func.TO_BASE64(Store.q.sample),
        ],
        sqlbuilder.AND(
            Store.q.name == name,
            sqlbuilder.func.FROM_UNIXTIME(Store.q.createdAt, DATE_FORMAT) >= sqlbuilder.func.FROM_UNIXTIME("2018-10-12", DATE_FORMAT)
        )
    )

    connection = Store._connection
    query = connection.sqlrepr(query)
    queryResult = connection.queryAll(query)
    return queryResult


def TO_BASE64(x):
    return x

def FROM_UNIXTIME(x, y):
    return sqlbuilder.SQLConstant('strftime("%Y%m%d", datetime({},"unixepoch", "localtime"))'.format(x))

for p in [
    mock.patch("sqlobject.sqlbuilder.func.TO_BASE64",TO_BASE64),
    mock.patch("sqlobject.sqlbuilder.func.FROM_UNIXTIME",FROM_UNIXTIME),
]:
    p.start()

retrieve('Some')

mock.patch.stopall()

产出如下：

 1/Query   :  CREATE TABLE store (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT,
    sample TEXT,
    created_at TIMESTAMP
)
 1/QueryR  :  CREATE TABLE store (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT,
    sample TEXT,
    created_at TIMESTAMP
)
 2/QueryAll:  SELECT store.sample FROM store WHERE (((store.name) = ('Some')) AND ((strftime("%Y%m%d", datetime(store.created_at,"unixepoch", "localtime"))) >= (strftime("%Y%m%d", datetime(2018-10-12,"unixepoch", "localtime")))))
 2/QueryR  :  SELECT store.sample FROM store WHERE (((store.name) = ('Some')) AND ((strftime("%Y%m%d", datetime(store.created_at,"unixepoch", "localtime"))) >= (strftime("%Y%m%d", datetime(2018-10-12,"unixepoch", "localtime")))))
 2/QueryAll-> []

PS。完全公开:我是SQLObject目前的维护者。

Answer 2

正如@phd指出的那样，SQLObject在将表达式以简化的形式传递到后端之前对其进行评估。

然后，我们也可以直接传递表达式，其中的SQLObject将计算，所以，而不是传递字符串文字，我们也可以这样做，如下

def FROM_UNIXTIME(x, y):
    return sqlbuilder.func.strftime("%Y%m%d", sqlbuilder.func.datetime(x, "unixepoch", "localtime"))

输出：

SELECT store.sample FROM store WHERE (((store.name) = ('Some')) AND ((strftime("%Y%m%d", datetime(store.created_at,"unixepoch", "localtime"))) >= (strftime("%Y%m%d", datetime(2018-10-12,"unixepoch", "localtime")))))