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

By simon at 2018-02-28 • 0人收藏 • 28人看过

在Django 2.0项目中,我在单元测试和I上遇到以下问题 找不到原因。 \ - 更新:我正在使用Postgre10.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())

4 个回复 | 最后更新于 2018-02-28
2018-02-28   #1

所以我通过测试进行调试,发现问题在这里发生。 您用于捕获请求的中间件不起作用在self.client.login。 因为它永远不会被调用。在你的第一个测试中你打电话

response = self.client.post(url, data=data)
这将调用中间件并设置它广告请求和当前用户。 但在你的下一个测试中,你有一个
investment = mommy.make(Investment, status=investment_status)
这将激发handle_status_track,然后获取较旧的请求是 从您之前的测试中剩余的并且具有id as的用户 1.但是现在的你ser与id=2,id=1 用户在测试1中创建并销毁。 所以你的中间件欺骗和ca该要求基本上是罪魁祸首 在这种情况下。 编辑-1 * 该问题只会在测试中发生,不会发生在生产。一个简单 修复以避免这是在中间件中创建set_user方法
def set_user(user):
    current_request = get_request()

    if current_request:
        current_request.user = user
然后更新你的login方法下面
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)
有一个尝试块,然后如何会发生异常吗?一行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,所以你得到错误的痕迹。不知道它是一个错误还是一个边缘 情况,但那是我的分析

2018-02-28   #2

我通过重构我的代码解决了这个问题。 现在我不会在save的投资方法中创建轨道 或插件ide是一个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信号导致此错误。

2018-02-28   #3

所以我通过测试进行调试,发现问题在这里发生。 您用于捕获请求的中间件不起作用在self.client.login。 因为它永远不会被调用。在你的第一个测试中你打电话

response = self.client.post(url, data=data)
这将调用中间件并设置它广告请求和当前用户。 但在你的下一个测试中,你有一个
investment = mommy.make(Investment, status=investment_status)
这将激发handle_status_track,然后获取较旧的请求是 从您之前的测试中剩余的并且具有id as的用户 1.但是现在的你ser与id=2,id=1 用户在测试1中创建并销毁。 所以你的中间件欺骗和ca该要求基本上是罪魁祸首 在这种情况下。 编辑-1 * 该问题只会在测试中发生,不会发生在生产。一个简单 修复以避免这是在中间件中创建set_user方法
def set_user(user):
    current_request = get_request()

    if current_request:
        current_request.user = user
然后更新你的login方法下面
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)
有一个尝试块,然后如何会发生异常吗?一行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,所以你得到错误的痕迹。不知道它是一个错误还是一个边缘 情况,但那是我的分析

2018-02-28   #4

正如@Tarun Lalwani所提到的那样,问题的根本原因很糟糕 请求中间件的管理 这里是固定的c请澄清:

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

登录后方可回帖

Loading...