如何在SQLAlchemy与alchemy-mock中Mock多表关联的跨表过滤查询？

问题背景 #

现有如下PostgreSQL Schema：

CREATE SCHEMA problem AUTHORIZATION blockchain;

CREATE TABLE problem.users (
    id serial PRIMARY KEY NOT NULL,
    nick varchar NOT NULL
);

CREATE TABLE problem.passwords_keys (
    id serial PRIMARY KEY NOT NULL,
    user_id serial references problem.users,
    password varchar NOT NULL,
    valid bool NOT null DEFAULT TRUE
);

通过sqlacodegen生成SQLAlchemy模型后，想要模拟执行带跨表过滤的查询：

SELECT p."password" "password" FROM 
    problem.passwords_keys p 
    JOIN problem.users u ON(p.user_id=u.id) 
    WHERE p."valid" = true AND u.nick = 'Bush';

使用UnifiedAlchemyMagicMock编写的mock代码在添加User.nick过滤条件后，返回None并报错AttributeError: 'NoneType' object has no attribute 'password'，但去掉跨表过滤后可以正常运行。

原因分析 #

mock-alchemy的UnifiedAlchemyMagicMock对跨表字段的filter条件匹配要求严格，手动定义的mock.call.filter(schema.User.nick == nick)无法和实际查询生成的SQLAlchemy表达式完全匹配，导致找不到对应的mock数据。另外，手动维护调用序列的方式灵活性差，不如让mock自动处理过滤逻辑可靠。

解决方案 #

方案1：预先注入关联测试数据，让mock自动处理过滤 #

无需手动定义查询调用序列，直接向mock session添加关联的User和PasswordsKey实例，mock-alchemy会自动处理join和filter逻辑：

from mock_alchemy.mocking import UnifiedAlchemyMagicMock
import schema

nick = 'Bush'

# 创建mock session
session = UnifiedAlchemyMagicMock()

# 构建关联的测试数据
test_user = schema.User(id=22, nick=nick)
test_password = schema.PasswordsKey(id=2, user_id=22, password='qwerty1', valid=True)

# 将数据加入mock session
session.add(test_user)
session.add(test_password)

# 执行目标查询
keys_for_user = session.query(schema.PasswordsKey)\
    .join(schema.User)\
    .filter(schema.PasswordsKey.valid == True)\
    .filter(schema.User.nick == nick)

password = keys_for_user.first().password
print(f'1. Found private key {password} by nick "{nick}"')

方案2：调整mock call的条件匹配方式（不推荐） #

如果必须手动定义call序列，需确保filter条件和实际查询生成的表达式完全一致：

from mock_alchemy.mocking import UnifiedAlchemyMagicMock
from unittest import mock
import schema

nick = 'Bush'

session = UnifiedAlchemyMagicMock(data=[
    (
        [
            mock.call.query(schema.PasswordsKey),
            mock.call.join(schema.User),
            mock.call.filter(schema.PasswordsKey.valid == True),
            mock.call.filter(schema.User.nick == nick),
        ],
        [schema.PasswordsKey(id=2, user_id=22, password='qwerty1', valid=True)]
    ),
])

keys_for_user = session.query(schema.PasswordsKey)\
    .join(schema.User)\
    .filter(schema.PasswordsKey.valid == True)\
    .filter(schema.User.nick == nick)

password = keys_for_user.first().password
print(f'1. Found private key {password} by nick "{nick}"')

说明 #

方案1是更推荐的实现方式，它模拟了真实的数据库查询逻辑，无需手动维护调用序列，当查询逻辑变化时，mock代码不需要同步修改，测试更健壮。

内容的提问来源于stack exchange，提问作者baziorek