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

Question

我正在编写涉及HTTP请求的测试。我必须按特定顺序提出一些请求，但我想检查每个步骤的响应。所以我认为生成器适合强制执行序列：# Main codedef sequence_of_requests(arg1, arg2):    yield request_a(arg1)    yield request_b(arg1, arg2)    yield request_c(arg1, arg2)然后在我的测试代码中，我可以写：# Test codegenerator_responses = sequence_of_requests()r = next(generator_responses)assert r.status_code == 200r = next(generator_responses)assert r.status_code == 204r = next(generator_responses)assert r.status_code == 404问题是，在第一次尝试时并不总是收到正确的状态，因此我一直在测试代码中使用装饰器包装此函数，以便它反复尝试，直到成功或超时。request_c()我想知道是否有某种方法可以在生成器上进行此包装，因此我仍然可以获得强制的事件序列。因此，我基本上想知道是否有某种方法可以反复调用生成器的一个屈服步长。注意：我不想将代码直接放在函数中等待响应，因为它不是测试代码。该函数仅用于确保以正确的顺序完成这些步骤。sequence_of_requestssequence_of_requests

Answer 1

您没有非常具体地说明超时和重试次数，或者成功的条件...但是，为简单起见，假设您希望重试最多 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

Answer 2

您可以生成函数，而不是生成响应值本身，然后可以重复调用这些函数。函数的参数可以使用 应用。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)