如何在生成器中重复执行一个屈服步骤？

Question

我正在编写涉及HTTP请求的测试。我必须按一定顺序提出一些要求，但我想检查每个步骤的响应。因此，我认为应该使用一个生成器来执行该序列：

def sequence_of_requests(arg1, arg2):

    yield request_a(arg1)
    yield request_b(arg1, arg2)
    yield request_c(arg1, arg2)

然后用我可以编写的测试代码

generator_responses = sequence_of_requests()

r = next(generator_responses)
assert r.status_code == 200

r = next(generator_responses)
assert r.status_code == 204

r = next(generator_responses)
assert r.status_code == 404

问题是request_c()在第一次尝试时并不总是获得正确的状态，因此我一直在测试代码中用装饰器包装此函数，因此它反复尝试直到成功或超时。

我想知道是否有某种方法可以在生成器上进行这种包装，所以我仍然得到强制的事件序列。因此，我基本上想知道是否有某种方法可以重复调用生成器的一个yield步骤。

注：我不想将代码直接放在sequence_of_requests函数中等待响应，因为它不是测试代码。 sequence_of_requests功能仅用于确保以正确的顺序完成这些步骤。

Answer 1

您可以产生函数，而不是产生响应值本身，然后可以重复调用这些函数。可以通过functools.partial应用功能的参数。

from functools import partial


def sequence_of_requests(arg1, arg2):

    yield partial(request_a, arg1)
    yield partial(request_b, arg1, arg2)
    yield partial(request_c, arg1, arg2)

现在可以测试步骤：

generator_functions = sequence_of_requests()

r = next(generator_functions)()
assert r.status_code == 200

r = next(generator_functions)()
assert r.status_code == 204

# Repeatedly make the request until the expected status code is returned.
func_slow_request = next(generator_functions)
wait_for_status(404)(func_slow_request)

Answer 2

您对超时和重试的次数或成功的条件不是很明确...但是，为简单起见，让我们假设您要重试最多5次，两次重试之间间隔1秒，并且您期望出现错误500（或5xx）以外的代码继续。

然后您可以使用类似：

import time

def sequence_of_requests(arg1, arg2):
    yield request_a(arg1)
    yield request_b(arg1, arg2)
    for retries in range(5):
        result = request_c(arg1, arg2)
        if result.status_code < 500:
            break
        time.sleep(1)
    yield result

如果达到最大重试次数，则只需返回最后收到的响应。如果得到有效的响应，则可以跳出循环并返回良好的响应。

效率不高，如果达到重试总数，在返回值之前，您仍然会睡1秒钟。您可以通过显式管理retries变量并检查循环内的限制来解决此问题。但是您可能对如何管理放弃有其他想法（例如超时，而不是固定的重试次数），因此您应该调整代码的这一部分，以遵循对您有意义的逻辑。