Один из способов перехватить изменения в базе данных - это использовать механизм расширения сеанса SQLAlchemy и перехватывать сброс к базе данных, используя что-то вроде этого:
from sqlalchemy.orm.attributes import instance_state
from sqlalchemy.orm import SessionExtension
class MockExtension(SessionExtension):
def __init__(self):
self.clear()
def clear(self):
self.updates = set()
self.inserts = set()
self.deletes = set()
def before_flush(self, session, flush_context, instances):
for obj in session.dirty:
self.updates.add(obj)
state = instance_state(obj)
state.commit_all({})
session.identity_map._mutable_attrs.discard(state)
session.identity_map._modified.discard(state)
for obj in session.deleted:
self.deletes.add(obj)
session.expunge(obj)
self.inserts.update(session.new)
session._new = {}
Затем для тестов вы можете настроить сеанс с этим макетом и посмотреть, соответствует ли он вашим ожиданиям.
mock = MockExtension()
Session = sessionmaker(extension=[mock], expire_on_commit=False)
def do_something(attr):
session = Session()
obj = session.query(Cls).first()
obj.attr = attr
session.commit()
def test_something():
mock.clear()
do_something('foobar')
assert len(mock.updates) == 1
updated_obj = mock.updates.pop()
assert updated_obj.attr == 'foobar'
Но вы все равно захотите провести хотя бы несколько тестов с базой данных, потому что по крайней мере захотите узнать, работают ли ваши запросы так, как ожидалось. И имейте в виду, что вы также можете вносить изменения в базу данных через session.update(), .delete() и .execute().