为何Django测试会打开大量数据库连接？

问题重现 #

测试代码如下：

from contextlib import contextmanager

import schemathesis
from hypothesis import given, settings
from hypothesis.extra.django import LiveServerTestCase as HypothesisLiveServerTestCase

from my_app.api import api_v1
from my_app.tests import AuthTokenFactory

# 3 requests per second - `3/s`
# 100 requests per minute - `100/m`
# 1000 requests per hour - `1000/h`
# 10000 requests per day - `10000/d`
RATE_LIMIT = "10/s"

openapi_schema = api_v1.get_openapi_schema()


@contextmanager
def authenticated_strategy(strategy, token: str):
    @given(case=strategy)
    @settings(deadline=None)
    def f(case):
        case.call_and_validate(headers={"Authorization": f"Bearer {token}"})

    yield f


class TestApiSchemathesis(HypothesisLiveServerTestCase):
    """
    Tests the REST API using Schemathesis.

    It does this by spinning up a live server, and then uses Schemathesis to
    automatically generate schema-conforming requests to the API.
    """

    def setUp(self):
        super().setUp()
        self.schema = schemathesis.from_dict(
            openapi_schema, base_url=self.live_server_url, rate_limit=RATE_LIMIT
        )

    def test_api(self):
        """
        Loop over all API endpoints and methods, and runs property tests for each.
        """
        auth_token = AuthTokenFactory().token
        for endpoint in self.schema:
            for method in self.schema[endpoint]:
                with self.subTest(endpoint=f"{method.upper()} {endpoint}"):
                    strategy = self.schema[endpoint][method].as_strategy()
                    with authenticated_strategy(strategy, auth_token) as run_strategy:
                        run_strategy()

新增端点后出现数据库连接耗尽错误：

psycopg2.OperationalError: connection to server at "localhost" (::1), port 5432 failed: FATAL: sorry, too many clients already

django.db.utils.OperationalError: database "test_myapp" is being accessed by other users
DETAIL: There are 99 other sessions using the database.

可能原因 #

• Hypothesis默认多线程生成测试用例，每个请求新建数据库连接且未及时释放
• LiveServerTestCase启动的后台线程未正确回收连接
• 批量遍历端点时，大量请求的连接累积未关闭

解决措施 #

1. 限制Hypothesis并发与测试用例数 #

修改authenticated_strategy中的@settings，禁用并发并减少单端点测试用例数，避免短时间内创建大量连接：

@contextmanager
def authenticated_strategy(strategy, token: str):
    @given(case=strategy)
    # 禁用多线程，限制单端点测试用例数为50（可根据需求调整）
    @settings(deadline=None, workers=1, max_examples=50)
    def f(case):
        case.call_and_validate(headers={"Authorization": f"Bearer {token}"})

    yield f

2. 手动强制回收数据库连接 #

在每个子测试结束后，手动关闭Django数据库连接，避免连接泄漏：

def test_api(self):
    auth_token = AuthTokenFactory().token
    from django.db import connections
    for endpoint in self.schema:
        for method in self.schema[endpoint]:
            with self.subTest(endpoint=f"{method.upper()} {endpoint}"):
                strategy = self.schema[endpoint][method].as_strategy()
                with authenticated_strategy(strategy, auth_token) as run_strategy:
                    run_strategy()
                # 关闭所有数据库连接
                for conn in connections.all():
                    conn.close()

同时在测试类的tearDown方法中补充连接清理：

def tearDown(self):
    super().tearDown()
    from django.db import connections
    for conn in connections.all():
        conn.close()

3. 降低Schemathesis速率限制 #

将RATE_LIMIT从10/s调低至5/s或3/s，减少单位时间内的请求量，让连接有足够时间被回收：

RATE_LIMIT = "5/s"

4. 改用WSGI模式替代LiveServerTestCase #

如果不需要真实HTTP服务器，直接通过WSGI调用应用，避免LiveServer额外的线程和连接开销：

from django.test import TestCase
from my_app.wsgi import application

class TestApiSchemathesis(TestCase):
    def setUp(self):
        super().setUp()
        self.schema = schemathesis.from_wsgi(
            openapi_schema, application, rate_limit=RATE_LIMIT
        )

5. 临时调整PostgreSQL最大连接数（应急方案） #

若CI环境允许，修改PostgreSQL配置postgresql.conf中的max_connections参数（如从100改为200），但此方法仅为临时缓解，优先修复代码层面问题。

调试方法 #

• 实时查看连接数：在测试过程中插入SQL查询，打印当前活跃连接数：

  from django.db import connection
  with connection.cursor() as cursor:
      cursor.execute("SELECT count(*) FROM pg_stat_activity;")
      print(f"Active connections: {cursor.fetchone()[0]}")
  

• 启用Django连接跟踪：在测试环境的settings.py中设置CONN_MAX_AGE=0，强制每次请求后关闭连接；开启DEBUG=True可查看连接的创建/关闭日志。
• Hypothesis verbose日志：在@settings中添加verbosity=2，查看每个测试用例的执行细节，确认是否存在并发请求导致的连接暴涨。

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