Как выполнить модульное тестирование потоковых генераторов - PullRequest
Новая
       6

Как выполнить модульное тестирование потоковых генераторов

0 голосов
/ 11 декабря 2018

Я настраиваю модульные тесты для своего приложения для колбМногие из моих функций передают вывод в графический интерфейс, как этот: 

def stream():
    def generate():
        if request.method == "POST":
            hostname = request.data.decode('utf-8')
            hostname_dn = "{}.{}".format(hostname, DOMAIN)

            logging.info("Connecting to: {}".format(hostname_dn))

            # Connect to hostname and execute create reports
            client = set_up_client()
            client.connect(hostname_dn,
                           username=USERNAME,
                           password=PASSWORD)

            cmd = ('tail -f -n0 /home/server.log')

            stdin, stdout, stderr = client.exec_command(cmd)

            for line in iter(lambda: stdout.readline(2048), ""):
                logging.info(line, end="")
                yield line
                if re.search(r'keyword', line):
                    yield 'keyword detected\n'
                    break

    return Response(stream_with_context(generate()), mimetype='text/html')

Мой вопрос: как я могу использовать операторы assert для проверки этих функций?Так как они возвращают ответ потока.Есть ли способ, которым я могу добавить дополнительный параметр в операторе возврата, например, 200 или что-то еще, затем использовать assert для проверки успешности потока?

1 Ответ

0 голосов
/ 12 декабря 2018

В вашем случае вы должны проверить, как работает stream функция.Поэтому я бы рекомендовал изолировать (смоделировать) все, что к нему не относится, и проверить, как оно себя ведет.Так что смоделируйте Response объект и итерируйте по генератору: 

@patch("Response")
def test_stream(self, response_mock):
    # this should return invoked Response mock, 
    # so you need to retrieve a first argument
    res = stream()
    args, _ = res.call_args
    stream_gen = args[0]
    n_runs = 0
    for i in stream_gen:
        self.assertEqual(i, expected_value)
        n_runs += 1
    self.assertEqual(n_runs, expected_runs_count)
...