等待 Popen 上的标准输出

Question

我正在尝试为烧瓶应用程序设置验收测试工具，目前我正在努力等待应用程序启动后再拨打电话。

以下构造工作正常：

class SpinUpTests(unittest.TestCase):
def tearDown(self):
    super().tearDown()
    self.stubby_server.kill()
    self.stubby_server.communicate()

def test_given_not_yet_running_when_created_without_config_then_started_on_default_port(self):
    self.not_yet_running(5000)

    self.stubby_server = subprocess.Popen(['python', '../../app/StubbyServer.py'], stdout=subprocess.PIPE)
    time.sleep(1)#<--- I would like to get rid of this

    self.then_started_on_port(5000)

我想在标准输出上等待：

self.stubby_server = subprocess.Popen(['python', '../../app/StubbyServer.py'], stdout=subprocess.PIPE) time.sleep(1)#

• 在http://127.0.0.1:[port]/ 上运行（按 CTRL+C 退出）

我试过了

for line in self.stubby_server.stdout.readline()

但是readline() 永远不会完成，尽管我已经在测试输出窗口中看到了输出。

有什么想法可以等待烧瓶应用程序启动而无需使用明确的sleep()？

Answer 1

使用retry 包，这将有助于解决您的问题。最终，您设置 what 您希望重试，what 您想要重试的异常，并且您可以根据您想要重试的方式设置特定的时序参数。它有很好的记录。

这是我如何在我正在从事的一个项目中解决此问题的示例here

如果链接不起作用，以下是可以帮助您的代码的 sn-p：

@classmethod
def _start_app_locally(cls):
    subprocess.Popen(["fake-ubersmith"])
    retry_call(
        requests.get,
        fargs=["{}/status".format(cls.endpoint)],
        exceptions=RequestException,
        delay=1
    )

如您所见，我只是尝试使用requests 使用get 访问我的端点（fargs 是传递给requests.get 的参数，您可以看到它调用back您传递给retry_call 的方法），并基于我期待的RequestException，我会延迟1 秒重试。

最后，“fake-ubersmith”是运行你的服务器的命令，它最终是你的类似命令：'python', '../../app/StubbyServer.py'