Django 应用程序：由于 django.db.utils.IntegrityError 单元测试失败

Question

在 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 单元测试显示了问题。

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

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())

Answer 1

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

您用于捕获请求的中间件在self.client.login. 因为它永远不会被调用。在你的第一次测试中你打电话

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

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

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

这会触发handle_status_track，然后它会获取之前测试中剩余的旧请求，并让用户使用idas 1。但当前用户是id=2，该id=1用户是在测试 1 本身中创建和销毁的。

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

编辑-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)

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

错误的异常堆栈跟踪

您的例外情况如下

  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块，那怎么会出现异常呢？其中第 188 行testcases.py，你有

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，因此您会得到错误的跟踪。不确定这是一个错误还是一个边缘情况，但我的分析