Django测试中time.sleep()导致PostgreSQL数据库时间冻结问题

问题描述 #

在使用pytest-django的Django项目测试中添加暂停操作时，time.sleep()虽能暂停代码执行，但出现异常副作用：PostgreSQL的"数据库时钟"被冻结，两次查询NOW()得到的时间完全一致，不符合预期。测试代码如下：

from pytest import mark, fixture, raises
from django.db import connection
from django.utils import timezone
import sys, datetime as dt
from time import sleep

@mark.django_db
def test_db_time ():
    print('Code time before:', timezone.now(), file = sys.stderr)
    with connection.cursor() as cursor:
        cursor.execute("SELECT NOW()")
        print('DB NOW() before:',
              t_before := cursor.fetchone()[0],
              file = sys.stderr)

    sleep(1.56)

    print('Code time after:', timezone.now(), file = sys.stderr)
    with connection.cursor() as cursor:
        cursor.execute("SELECT NOW()")
        print('DB NOW() after:',
              t_after := cursor.fetchone()[0],
              file = sys.stderr)

    assert t_before != t_after, 'I need a nap!'

执行测试后的输出：

========================================= test session starts =========================================
platform linux -- Python 3.10.1, pytest-7.4.0, pluggy-1.0.0
django: settings: pms.settings (from ini)
rootdir: /home/nikita/files/projects/pms/dev
configfile: pyproject.toml
plugins: django-4.5.2
collected 1 item                                                                                       

sis/tests/db/triggers/test_answer.py F                                                          [100%]

============================================== FAILURES ===============================================
____________________________________________ test_db_time _____________________________________________

    def test_db_time ():
        print('Code time before:', timezone.now(), file = sys.stderr)
        with connection.cursor() as cursor:
            cursor.execute("SELECT NOW()")
            print('DB NOW() before:',
                  t_before := cursor.fetchone()[0],
                  file = sys.stderr)
    
        sleep(1.56)
    
        print('Code time after:', timezone.now(), file = sys.stderr)
        with connection.cursor() as cursor:
            cursor.execute("SELECT NOW()")
            print('DB NOW() after:',
                  t_after := cursor.fetchone()[0],
                  file = sys.stderr)
    
>       assert t_before != t_after, 'I need a nap!'
E       AssertionError: I need a nap!
E       assert datetime.datetime(2023, 9, 7, 15, 14, 14, 414343, tzinfo=datetime.timezone.utc) != datetime.datetime(2023, 9, 7, 15, 14, 14, 414343, tzinfo=datetime.timezone.utc)

sis/tests/db/triggers/test_answer.py:114: AssertionError
---------------------------------------- Captured stderr call -----------------------------------------
Code time before: 2023-09-07 15:14:14.414191+00:00
DB NOW() before: 2023-09-07 15:14:14.414343+00:00
Code time after: 2023-09-07 15:14:15.976308+00:00
DB NOW() after: 2023-09-07 15:14:14.414343+00:00
======================================= short test summary info =======================================
FAILED sis/tests/db/triggers/test_answer.py::test_db_time - AssertionError: I need a nap!
========================================== 1 failed in 3.05s ==========================================

原本以为NOW()应该获取系统时间，但sleep期间系统时间正常运行，为何会出现这种现象？如何让数据库时钟正常工作？

问题原因 #

1. PostgreSQL的NOW()特性：PostgreSQL中NOW()（等价于transaction_timestamp()）返回的是当前事务开始的时间，而非实时系统时间。只要处于同一个事务内，无论何时调用NOW()，都会返回事务启动时的时间戳。
2. pytest-django的事务默认行为：@mark.django_db装饰器默认会在测试函数启动时自动开启一个数据库事务，测试结束后回滚事务（避免测试数据污染）。整个测试的所有数据库操作都在这同一个事务中，因此两次NOW()查询得到的时间完全一致。

解决方法 #

方法1：使用clock_timestamp()获取实时时间 #

如果仅需获取数据库服务器的实时系统时间，直接将NOW()替换为clock_timestamp()即可，该函数返回调用时的实际系统时间：

cursor.execute("SELECT clock_timestamp()")

方法2：手动控制事务边界 #

在sleep前后手动提交并重新开启事务，让两次数据库查询处于不同事务中：

from pytest import mark, fixture, raises
from django.db import connection, transaction
from django.utils import timezone
import sys, datetime as dt
from time import sleep

@mark.django_db
def test_db_time():
    print('Code time before:', timezone.now(), file = sys.stderr)
    with connection.cursor() as cursor:
        cursor.execute("SELECT NOW()")
        print('DB NOW() before:',
              t_before := cursor.fetchone()[0],
              file = sys.stderr)
    # 提交当前事务
    transaction.commit()

    sleep(1.56)

    # 重新开启事务
    transaction.begin()
    print('Code time after:', timezone.now(), file = sys.stderr)
    with connection.cursor() as cursor:
        cursor.execute("SELECT NOW()")
        print('DB NOW() after:',
              t_after := cursor.fetchone()[0],
              file = sys.stderr)

    assert t_before != t_after, 'I need a nap!'

方法3：禁用自动事务管理 #

使用@mark.django_db(transaction=False)装饰器关闭测试的自动事务功能，此时每个数据库操作都会自动提交：

@mark.django_db(transaction=False)
def test_db_time():
    # 原测试代码不变

注意：该方式会让测试中的数据库操作直接写入数据库，测试后不会自动回滚，需自行清理测试数据，避免影响其他测试用例。

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