Совместный запуск теста не удался, запуск теста по отдельности удался

Я использую pytest.

Пройдя через тесты с отладчиком, запустив оба test_joining_previously_entered_queue_returns_previous_queue_details и test_joining_queue_enters_correct_position вместе, test_joining_previously_entered_queue_returns_previous_queue_details проходит успешно, но test_joining_queue_enters_correct_position терпит неудачу на queue = get_object_or_404(Queue, pk=kwargs["pk"]) в представлении JoinQueue, где представление бросает 404

Если я запускаю test_joining_previously_entered_queue_returns_previous_queue_details по отдельности, то тест проходит без проблем.

Должно произойти так: create_company_one создает объект Company, который в свою очередь создает объект Queue (показанный в переопределенном методе save в Company). Связанный объект Queue не создается, если test_joining_queue_enters_correct_position выполняется вместе с другим тестом, но работает, если тест выполняется отдельно

Почему это происходит и как я могу это исправить?

test_join_queue.py

def create_company_one():

    return Company.objects.create(name='kfc')

@pytest.mark.django_db
def test_joining_previously_entered_queue_returns_previous_queue_details(authenticated_client: APIClient):
    
    create_company_one()
    
    url = reverse('user-queue-join', args=[1])
    
    first_response = authenticated_client.put(url)
    
    first_data = first_response.data
    
    second_response = authenticated_client.put(url)
    
    second_data = second_response.data
    
    assert first_data == second_data

@pytest.mark.django_db
def test_joining_queue_enters_correct_position(factory: APIRequestFactory):
    """
    Makes sure that whenever a user joins a queue, their position is sorted correctly
    based on WHEN they joined. Since all users in self.users join the queue
    in a time canonical manner, we can iterate over them to test for their queue positions.
    """
    
    create_company_one()
    
    users = create_users()
    view = JoinQueue.as_view()
    url = reverse('user-queue-join',args=[1])
    request = factory.put(url)
    for queue_position, user in enumerate(users):
        
        force_authenticate(request, user=user)
        response = view(request, pk=1)
        assert response.data["position"] == queue_position + 1

views.py

class JoinQueue(APIView):

    permission_classes = [IsAuthenticated]

    def put(self, request, *args, **kwargs):

        # throws 404 here

        queue = get_object_or_404(Queue, pk=kwargs["pk"])

        user = request.user

        try:
            queue_details_of_a_user = user.queue_details.get(queue=queue)
        except ObjectDoesNotExist:
            queue_details_of_a_user = QueueDetails.objects.create(user=user, queue=queue)
            serializer = QueueDetailsSerializer(queue_details_of_a_user)
            return Response(serializer.data)
        else:
            serializer = QueueDetailsSerializer(queue_details_of_a_user)
            return Response(serializer.data)

models.py

class Company(models.Model):
    name = models.CharField(max_length=15, unique=True)
    
    def save(self, *args: Any, **kwargs: Any) -> None:
        created = bool(self.pk)
        super().save(*args, **kwargs)
        if not created:
            Queue.objects.create(company=self)
            logging.info(f"{self.name} has created its queue")

        return None

    def __str__(self) -> str:
        return self.name

class Queue(models.Model):
    company = models.OneToOneField(
        Company, on_delete=models.CASCADE, related_name="queue"
    )
    users = models.ManyToManyField(User, through="QueueDetails", related_name="queues")

    @property
    def length(self) -> int:
        return len(self.users.all())

    @property
    def sorted_users(self) -> "QuerySet[User]":
        return self.users.all().order_by("queue_details__joined_at")

    def __str__(self) -> str:
        return f"{self.company}'s queue"


class QueueDetails(models.Model):
    
    joined_at = models.DateTimeField(auto_now_add=True)
    queue = models.ForeignKey(
        Queue,
        related_name="queue_details",
        on_delete=models.CASCADE,
    )
    user = models.ForeignKey(
        User,
        related_name="queue_details",
        on_delete=models.CASCADE,
    )

    @property
    def position(self) -> int:
        for index, user in enumerate(self.queue.sorted_users):
            if user == self.user:
                return index + 1
        raise ValueError("User is not in the queue, Invalid queue position")

Вернуться на верх