如何在 Django 测试中发送命令行输入响应

Question

我有一个 Django 管理命令，它会要求输入命令行 (y/n)，我现在正在为此编写一个测试。

目前，当我运行测试时（就像现在一样），测试将停止并等待 y/n 输入。但是如何编写/更新我的测试，以便向 call_command 函数发送 y/n 响应？

关于信息，我的管理命令代码是（已删除导入 - 但完整代码位于：https://github.com/DigitalCampus/django-oppia/blob/master/oppia/management/commands/remove_duplicate_trackers.py）：

class Command(BaseCommand):
    help = _(u"Removes any duplicate trackers based on UUID")

    def handle(self, *args, **options):
        """
        Remove page/media/quiz trackers with no UUID
        """
        result = Tracker.objects.filter(Q(type='page')
                                        | Q(type='quiz')
                                        | Q(type='media'),
                                        uuid=None).delete()
        print(_(u"\n\n%d trackers removed that had no UUID\n" % result[0]))

        """
        Remove proper duplicate trackers - using min id
        """
        trackers = Tracker.objects.filter(Q(type='page')
                                          | Q(type='quiz')
                                          | Q(type='media')) \
            .values('uuid') \
            .annotate(dcount=Count('uuid')) \
            .filter(dcount__gte=2)

        for index, tracker in enumerate(trackers):
            print("%d/%d" % (index, trackers.count()))
            exclude = Tracker.objects.filter(uuid=tracker['uuid']) \
                .aggregate(min_id=Min('id'))
            deleted = Tracker.objects.filter(uuid=tracker['uuid']) \
                .exclude(id=exclude['min_id']).delete()
            print(_(u"%d duplicate tracker(s) removed for UUID %s based on \
                   min id" % (deleted[0], tracker['uuid'])))

        """
        Remember to run summary cron from start
        """
        if result[0] + trackers.count() > 0:
            print(_(u"Since duplicates have been found and removed, you \
                    should now run `update_summaries` to ensure the \
                    dashboard graphs are accurate."))
            accept = input(_(u"Would you like to run `update_summaries` \
                                now? [Yes/No]"))
            if accept == 'y':
                call_command('update_summaries', fromstart=True)

我的测试代码是：

def test_remove_with_duplicates(self):
    Tracker.objects.create(
        user_id=1,
        course_id = 1,
        type = "page",
        completed = True,
        time_taken = 280,
        activity_title = "{\"en\": \"Calculating the uptake of antenatal care services\"}",
        section_title = "{\"en\": \"Planning Antenatal Care\"}",
        uuid = "835713f3-b85e-4960-9cdf-128f04014178")
    out = StringIO()
    tracker_count_start = Tracker.objects.all().count()

    call_command('remove_duplicate_trackers', stdout=out)

    tracker_count_end = Tracker.objects.all().count()
    self.assertEqual(tracker_count_start-1, tracker_count_end)

非常感谢任何帮助，如果您需要任何额外的信息/代码，请告诉我，谢谢。

编辑 我尝试了@xyres 的建议来添加“interactive=False”，但我收到了一个类型错误：

TypeError: Unknown option(s) for remove_duplicate_trackers command: interactive. Valid options are: force_color, help, no_color, pythonpath, settings, skip_checks, stderr, stdout, traceback, verbosity, version

然后我也尝试使用 'skip_checks=True' 但这仍然使测试挂起等待命令行输入

Answer 1

我喜欢使用mock.patch 来处理这类事情。这是一个基于您的命令的精简示例。一、只提示并接受一行输入的管理命令

from django.core.management import BaseCommand


class Command(BaseCommand):

    def handle(self, *args, **options):
        accept = input("Would you like to run `update_summaries` now? [Yes/No]")
        if accept == 'y':
            self.stdout.write("Updating Summaries!")
            return
        self.stdout.write("Not updating Summaries...")

接下来，这里是TestCase。我喜欢使用调用包装器来整合修补逻辑并允许测试不同的选项等。

from io import StringIO

from unittest import mock

from django.core.management import call_command

from django.test import TestCase


class UpdateTestCase(TestCase):

    @mock.patch("my_app.management.commands.update.input")
    def _call_wrapper(self, response_value, mock_input=None):
        def input_response(message):
            return response_value
        mock_input.side_effect = input_response
        out = StringIO()
        call_command('update', stdout=out)
        return out.getvalue().rstrip()

    def test_yes(self):
        """Test update command with "y" response
        """
        self.assertEqual("Updating Summaries!", self._call_wrapper('y'))

    def test_no(self):
        """Test update command with "n" response
        """
        self.assertEqual("Not updating Summaries...", self._call_wrapper('n'))