使用 before_after 进行 Django 竞争条件测试

Question

我已经完成了 Django polls 教程，我想通过测试来检查 here 中提到的投票竞争条件。我发现了一个名为before_after 的python 包，可用于进行涉及竞争条件的测试。但是我不明白before 和after 方法的参数。我查看了some examples 并认为Choice.get_votes 是我的第一个参数所需要的，但这导致我得到ImportError: No module named 'Choice'。有谁知道我需要在测试中进行哪些更改才能使其正常工作？

这是我做的测试：

tests.py：

import datetime
from django.utils import timezone
from django.test import TestCase
from .models import Question, Choice
import before_after

def create_question(question_text, days):
    time = timezone.now() + datetime.timedelta(days=days)
    return Question.objects.create(question_text=question_text, pub_date=time)

class ChoiceTest(TestCase):

    def test_race_condition_avoided(self):
        self.assertEqual(self.creating_race_condition(), 2, "Votes do not equal amount expected.")

    def test_race_condition_occurred(self):
        self.assertNotEqual(self.creating_race_condition(), 1, "Votes do equal amount expected from a race condition.")

    def creating_race_condition(self):
        q = create_question('Did the race condition happen?', 0)
        self.choice = q.choice_set.create(choice_text="Let's find out.", votes=0)
        with before_after.after('Choice.get_votes', self.increment_vote):
            self.increment_vote()
        return c.votes

    def increment_vote(self):
        self.choice.votes += 1
        self.choice.save()

    def get_votes(self):
        return self.choice.votes

models.py：

import datetime
from django.utils import timezone
from django.db import models

class Question(models.Model):
    question_text = models.CharField(max_length=200)
    pub_date = models.DateTimeField('date published')

    def __str__(self):
        return self.question_text

    def was_published_recently(self):
        now = timezone.now()
        return now - datetime.timedelta(days=1) <= self.pub_date <= now
    was_published_recently.admin_order_field = 'pub_date'
    was_published_recently.boolean = True
    was_published_recently.short_description = 'Published recently?'

class Choice(models.Model):
    question = models.ForeignKey(Question, on_delete=models.CASCADE)
    choice_text = models.CharField(max_length=200)
    votes = models.IntegerField(default=0)

    def __str__(self):
        return self.choice_text

我正在使用 Python 3.5.1 和 Django 1.10.4

Answer 1

before_after 需要你使用完整路径来导入函数。

你需要像'vote.models.Choice.get_votes'这样使用str来给它完整的导入路径。

Answer 2

before-after 包的使用有点违反直觉。这是我的解决方案。

polls/models.py - 模型实现被忽略。

class Question(models.Model):
    ...

class Choice(models.Model):
    ...

polls/views.py - 为了缩短代码，我在这里有意删除了错误处理。希望日志有助于更好地演示。

def vote(request, question_id):
    print('REQUEST STARTED')
    question = get_object_or_404(Question, pk=question_id)
    print('DB QUERIED')
    selected_choice = question.choice_set.get(pk=request.POST['choice'])
    # use the F() expressions to avoid race condition
    selected_choice.votes = F('votes') + 1
    selected_choice.save()
    print('DB UPDATED')
    return HttpResponseRedirect(reverse('polls:results', args=(question.id,)))

polls/tests.py

class QuestionVoteViewTests(TestCase):
    def test_vote_increment_race(self):
        """
        When there're multiple votes increments to the same choice
        happening simultaneously,
        the value should be updated accordingly.
        """
        # the helper `create_question` returns a tuple
        # containing the question obj & a list of
        # the passing choices obj
        (past_question, [choice]) = create_question(
            question_text='Past Question.',
            days=-5,
            choices=['choice'],
        )
        # create a request instance here
        # to test the view function without using the test client
        request = HttpRequest()
        request.METHOD = 'POST'
        request.POST['choice'] = choice.id
        # here's the trickiest thing.
        # 1. the first argument should be the full path of the module,
        #    polls/models.py:Choice
        # 2. I fail here creating the second thread
        #    right after the first thread's DB query
        #    `polls.models.Question.choice_set.get` doesn't work
        #    so instead I create the second thread before updating the DB,
        #    and it works.
        with before_after.before(
            'polls.models.Choice.save',
            lambda _: vote(request, past_question.id),
        ):
            vote(request, past_question.id)
        self.assertEqual(Choice.objects.get(pk=choice.id).votes, 2)

运行测试时，日志将是：

• 请求已开始
• 已查询数据库
• 请求已开始
• 已查询数据库
• 数据库已更新
• 数据库已更新

相关博客链接：http://www.c-oreills.com/2015/03/01/testing-race-conditions-in-python.html