我正在开发一个Django应用程序(Django v3.2.10pytest v7.0.1pytest-django v4.5.2),它使用cursor对我的第二个DB:my_db2执行原始查询,但在运行测试时,所有查询都返回空结果,就像它们在并行事务上运行一样.
我的测试文件:
@pytest.mark.django_db(transaction=True, databases=['default', 'my_db2'])
class TestItems:
def test_people(self):
person1 = PeopleFactory() # Adds 1 person to my_db2
assert fetch_all_persons() == 1 # Fails Returns 0
我的工厂:
class PeopleFactory(factory.django.DjangoModelFactory):
id = factory.Sequence(lambda x: x + 1)
name = factory.Faker('first_name')
class Meta:
model = People
我的职能:
from django.db import connections
def fetch_all_persons():
with connections['my_db2'].cursor() as cursor:
cursor.execute(f"SELECT * FROM Persons")
return len(list(cursor.fetchall())):
According documentation transaction=True
应该可以防止这个问题,但事实并非如此,有人知道如何解决它吗?
注.-使用ORM不是一个选项,这只是一个简单的例子来说明这个问题.实际使用的查询要复杂得多.