Django 应用程序：单元测试因 django.db.utils.IntegrityError 而失败答案

【问题标题】：Django app : unit tests fails because of django.db.utils.IntegrityErrorDjango 应用程序：单元测试因 django.db.utils.IntegrityError 而失败
【发布时间】：2018-07-27 15:32:22
【问题描述】：

在 Django 2.0 项目中，我在单元测试中遇到以下问题，但找不到原因。

-- 更新：我使用的是 Postgres 10.1。当我切换到 sqlite3 时不会出现问题

我正在实现一个模型，它可以跟踪另一个模型的任何变化

from django.db import models
from django.contrib.auth.models import User
from django.db.models.signals import post_save
from django.dispatch import receiver

class Investment(models.Model):
     """the main model"""
     status = models.IntegerField()


class InvestmentStatusTrack(models.Model):
    """track every change of status on an investment"""
    investment = models.ForeignKey(Investment, on_delete=models.CASCADE)
    status = models.IntegerField()
    modified_on = models.DateTimeField(
        blank=True, null=True, default=None, verbose_name=_('modified on'), db_index=True
    )
    modified_by = models.ForeignKey(
        User, blank=True, null=True, default=None, verbose_name=_('modified by'), on_delete=models.CASCADE
    )

    class Meta:
        ordering = ('-modified_on', )

    def __str__(self):
        return '{0} - {1}'.format(self.investment, self.status)


@receiver(post_save, sender=Investment)
def handle_status_track(sender, instance, created, **kwargs):
    """add a new track every time the investment status change"""
    request = get_request()  # a way to get the current request
    modified_by = None
    if request and request.user and request.user.is_authenticated:
        modified_by = request.user
    InvestmentStatusTrack.objects.create(
       investment=instance, status=instance.status, modified_on=datetime.now(), modified_by=modified_by
    )

我的大部分单元测试都因以下回溯而失败

Traceback (most recent call last):
  File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 209, in __call__
    self._post_teardown()
  File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 893, in _post_teardown
    self._fixture_teardown()
  File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 1041, in _fixture_teardown
    connections[db_name].check_constraints()
  File "/env/lib/python3.6/site-packages/django/db/backends/postgresql/base.py", line 235, in check_constraints
    self.cursor().execute('SET CONSTRAINTS ALL IMMEDIATE')
  File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 68, in execute
    return self._execute_with_wrappers(sql, params, many=False, executor=self._execute)
  File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 77, in _execute_with_wrappers
    return executor(sql, params, many, context)
  File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 85, in _execute
    return self.cursor.execute(sql, params)
  File "/env/lib/python3.6/site-packages/django/db/utils.py", line 89, in __exit__
    raise dj_exc_value.with_traceback(traceback) from exc_value
  File "/env/lib/python3.6/site-packages/django/db/backends/utils.py", line 83, in _execute
    return self.cursor.execute(sql)
django.db.utils.IntegrityError: insert or update on table "investments_investmentstatustrack" violates foreign key constraint "investments_investme_modified_by_id_3a12fb21_fk_auth_user"
DETAIL:  Key (modified_by_id)=(1) is not present in table "auth_user".

任何想法，如何解决这个问题？

-- 更新：2 个单元测试显示了问题。

单独执行时两者都成功。似乎问题发生在单元测试拆解上。外键约束此时失败，因为用户已被删除。

class TrackInvestmentStatusTest(ApiTestCase):

    def login(self, is_staff=False):
        password = "abc123"
        self.user = mommy.make(User, is_staff=is_staff, is_active=True)
        self.user.set_password(password)
        self.user.save()
        self.assertTrue(self.client.login(username=self.user.username, password=password))

    def test_add_investment(self):
        """it should add a new investment and add a track"""
        self.login()

        url = reverse('investments:investments-list')

        data = {}

        response = self.client.post(url, data=data)

        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

        self.assertEqual(1, Investment.objects.count())
        investment = Investment.objects.all()[0]
        self.assertEqual(investment.status, Investment.STATUS_IN_PROJECT)

        self.assertEqual(1, InvestmentStatusTrack.objects.count())
        track = InvestmentStatusTrack.objects.all()[0]
        self.assertEqual(track.status, investment.status)
        self.assertEqual(track.investment, investment)
        self.assertEqual(track.modified_by, self.user)
        self.assertEqual(track.modified_on.date(), date.today())

    def test_save_status(self):
        """it should modify the investment and add a track"""

        self.login()

        investment_status = Investment.STATUS_IN_PROJECT

        investment = mommy.make(Investment, asset=asset, status=investment_status)
        investment_id = investment.id

        self.assertEqual(1, InvestmentStatusTrack.objects.count())
        track = InvestmentStatusTrack.objects.all()[0]
        self.assertEqual(track.status, investment.status)
        self.assertEqual(track.investment, investment)
        self.assertEqual(track.modified_by, None)
        self.assertEqual(track.modified_on.date(), date.today())

        url = reverse('investments:investments-detail', args=[investment.id])

        data = {
            'status': Investment.STATUS_ACCEPTED
        }

        response = self.client.patch(url, data=data)

        self.assertEqual(response.status_code, status.HTTP_200_OK)

        self.assertEqual(1, Investment.objects.count())
        investment = Investment.objects.all()[0]
        self.assertEqual(investment.id, investment_id)
        self.assertEqual(investment.status, Investment.STATUS_ACCEPTED)

        self.assertEqual(2, InvestmentStatusTrack.objects.count())
        track = InvestmentStatusTrack.objects.all()[0]
        self.assertEqual(track.status, Investment.STATUS_ACCEPTED)
        self.assertEqual(track.investment, investment)
        self.assertEqual(track.modified_by, self.user)
        self.assertEqual(track.modified_on.date(), date.today())

        track = InvestmentStatusTrack.objects.all()[1]
        self.assertEqual(track.status, Investment.STATUS_IN_PROJECT)
        self.assertEqual(track.investment, investment)
        self.assertEqual(track.modified_by, None)
        self.assertEqual(track.modified_on.date(), date.today())

【问题讨论】：

你做了所有的 Makemigrations/Migrates 吗？
是的，它与模型是最新的
要么我是盲人，要么你的代码似乎没问题。您能否添加一个抛出此 Traceback 的单元测试？
@Sativa 我已经用单元测试更新了我的代码。
一些数据库驱动程序的问题是，在发生 IntegrityError 之后，连接保持在无效状态，之后的每个查询也会失败。为了安全起见，您可能应该在拆卸步骤中发出transaction.rollback。

标签： python django postgresql

【解决方案1】：

所以我通过测试进行了调试，发现问题发生在这里。

您用于捕获请求的中间件在self.client.login 中不起作用。因为它永远不会被调用。在您的第一个测试中，您调用

response = self.client.post(url, data=data)

这会调用中间件并设置线程请求和当前用户。但是在你的下一个测试中，你有一个

investment = mommy.make(Investment, status=investment_status)

这会触发handle_status_track，然后它会获取先前测试遗留下来的旧请求，并将id 的用户设置为1。但是当前用户是id=2，id=1 用户是在测试 1 本身中创建和销毁的。

所以在这种情况下，你的中间件欺骗和捕获请求基本上是罪魁祸首。

Edit-1

该问题只会发生在测试中，不会发生在生产中。避免这种情况的一个简单解决方法是在中间件中创建 set_user 方法

def set_user(user):
    current_request = get_request()

    if current_request:
        current_request.user = user

然后将您的登录方法更新到下面

def login(self, is_staff=False):
    password = "abc123"
    self.user = mommy.make(User, is_staff=is_staff, is_active=True)
    self.user.set_password(password)
    self.user.save()
    self.assertTrue(self.client.login(username=self.user.username, password=password))
    set_user(self.user)

这将确保每个测试都获得正确的中间件。

错误的异常堆栈跟踪

你的例外有以下 lin

  File "/env/lib/python3.6/site-packages/django/test/testcases.py", line 1041, in _fixture_teardown
    connections[db_name].check_constraints()

现在如果你看那一行的代码

def _fixture_teardown(self):
    if not connections_support_transactions():
        return super()._fixture_teardown()
    try:
        for db_name in reversed(self._databases_names()):
            if self._should_check_constraints(connections[db_name]):
                connections[db_name].check_constraints()
    finally:
        self._rollback_atomics(self.atomics)

有一个try块，那怎么会出现异常呢？ testcases.py的188行，你有

def __call__(self, result=None):
    """
    Wrapper around default __call__ method to perform common Django test
    set up. This means that user-defined Test Cases aren't required to
    include a call to super().setUp().
    """
    testMethod = getattr(self, self._testMethodName)
    skipped = (
        getattr(self.__class__, "__unittest_skip__", False) or
        getattr(testMethod, "__unittest_skip__", False)
    )

    if not skipped:
        try:
            self._pre_setup()
        except Exception:
            result.addError(self, sys.exc_info())
            return
    super().__call__(result)
    if not skipped:
        try:
            self._post_teardown()
        except Exception:
            result.addError(self, sys.exc_info())
            return

result.addError(self, sys.exc_info()) 捕获已由self._post_teardown 处理的异常，因此您得到错误的跟踪。不确定这是错误还是边缘情况，但我的分析

【讨论】：

你是绝对正确的。我不认为请求中间件可能是根本原因。您认为这只是一个单元测试问题还是应用程序中也会发生？
我已经通过在处理完请求后清除请求来修复中间件。我认为这主要是一个单元测试问题，但在某些情况下也可能是一个问题。

【解决方案2】：

我通过重构代码解决了这个问题。

现在我不在投资的 save 方法或 post_save 信号处理程序内创建轨道，而是在显式调用的方法中创建轨道

我的代码如下：

models.py

class Investment(models.Model):
    """the main model"""
    status = models.IntegerField()

    def handle_status_track(self):
        """add a new track every time the investment status change"""
        request = get_request()  # a way to get the current request
        modified_by = None
        if request and request.user and request.user.is_authenticated:
            modified_by = request.user
        InvestmentStatusTrack.objects.create(
            investment=self, status=self.status, modified_on=datetime.now(), modified_by=modified_by
        )


class InvestmentStatusTrack(models.Model):
    """track every change of status on an investment"""
    investment = models.ForeignKey(Investment, on_delete=models.CASCADE)
    status = models.IntegerField()
    modified_on = models.DateTimeField(
        blank=True, null=True, default=None, verbose_name=_('modified on'), db_index=True
    )
    modified_by = models.ForeignKey(
        User, blank=True, null=True, default=None, verbose_name=_('modified by'), on_delete=models.CASCADE
    )

    class Meta:
        ordering = ('-modified_on',)

views.py

class InvestmentViewSet(ViewSet):
    model = Investment
    serializer_class = InvestmentSerializer

    def perform_create(self, serializer):
        """save"""
        investment = serializer.save()
        investment.handle_status_track()

    def perform_update(self, serializer):
        """save"""
        investment = serializer.save()
        investment.handle_status_track()

问题在于它的作用并不完全相同：我需要在保存对象的任何时候处理对方法的调用。我还在想为什么post_save 信号会导致这个错误。

【讨论】：

不知道它是否相关，但取决于您的设置如何发出信号，我必须在运行单元测试时输入 weak=False(docs.djangoproject.com/en/2.0/topics/signals/…) 以避免它们被垃圾收集并导致奇怪的问题。
@DanielBackman 感谢您提供此信息。我不知道这个。我会检查的。
@DanielBackman谢谢你的建议。不幸的是，在设置信号时输入weak=False并没有解决问题
@luc，我发现您的测试存在问题，以及使用自定义中间件使用get_request 会污染系统状态

【解决方案3】：

正如@Tarun Lalwani 所提到的，问题的根本原因是对请求中间件的管理不善

这里是澄清的固定代码：

from threading import current_thread

class RequestManager(object):
    """get django request from anywhere"""
    _shared = {}

    def __init__(self):
        """This is a Borg"""
        self.__dict__ = RequestManager._shared

    def _get_request_dict(self):
        """request dict"""
        if not hasattr(self, '_request'):
            self._request = {}  # pylint: disable=attribute-defined-outside-init
        return self._request

    def clean(self):
        """clean"""
        if hasattr(self, '_request'):
            del self._request

    def get_request(self):
        """return request"""
        _requests = self._get_request_dict()
        the_thread = current_thread()
        if the_thread not in _requests:
            return None
        return _requests[the_thread]

    def set_request(self, request):
        """set request"""
        _requests = self._get_request_dict()
        _requests[current_thread()] = request


class RequestMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Set the request
        RequestManager().set_request(request)

        response = self.get_response(request)

        # ---- THIS WAS THE MISSING PART -----
        # Clear the request
        RequestManager().set_request(None)
        # ------------------------------------

        return response

    def process_exception(self, request, exception):
        """handle exceptions"""
        # clear request also in case of exception
        RequestManager().set_request(None)


def get_request():
    """get current request from anywhere"""
    return RequestManager().get_request()

【讨论】：