RobotFramework 与 Python 的 asyncio

Question

我正在尝试使用 Python3.6 的 asyncio 运行 RobotFramework。

相关的 Python-Code 如下所示：

""" SampleProtTest.py """

import asyncio
import threading

class SubscriberClientProtocol(asyncio.Protocol):
    """
    Generic, Asynchronous protocol that allows sending using a synchronous accessible queue
    Based on http://stackoverflow.com/a/30940625/4150378
    """
    def __init__(self, loop):
        self.loop = loop

    """ Functions follow for reading... """


class PropHost:
    def __init__(self, ip: str, port: int = 50505) -> None:
        self.loop = asyncio.get_event_loop()
        self.__coro = self.loop.create_connection(lambda: SubscriberClientProtocol(self.loop), ip, port)
        _, self.__proto = self.loop.run_until_complete(self.__coro)
        # run the asyncio-loop in background thread
        threading.Thread(target=self.runfunc).start()

    def runfunc(self) -> None:
        self.loop.run_forever()

    def dosomething(self):
        print("I'm doing something")


class SampleProtTest(object):
    def __init__(self, ip='127.0.0.1', port=8000):
        self._myhost = PropHost(ip, port)

    def do_something(self):
        self._myhost.dosomething()

if __name__=="__main__":
    tester = SampleProtTest()
    tester.do_something()

如果我在 python 中运行这个文件，它会按预期打印：

I'm doing something

为了在 Robot-Framework 中运行代码，我编写了以下 .robot 文件：

*** Settings ***
Documentation     Just A Sample
Library           SampleProtTest.py
*** Test Cases ***
Do anything
    do_something

但如果我运行这个 .robot 文件，我会收到以下错误：

Initializing test library 'SampleProtTest' with no arguments failed: This event loop is already running
Traceback (most recent call last):
  File "SampleProtTest.py", line 34, in __init__
    self._myhost = PropHost(ip, port)
  File "SampleProtTest.py", line 21, in __init__
    _, self.__proto = self.loop.run_until_complete(self.__coro)
  File "appdata\local\programs\python\python36\lib\asyncio\base_events.py", line 454, in run_until_complete
    self.run_forever()
  File "appdata\local\programs\python\python36\lib\asyncio\base_events.py", line 408, in run_forever
    raise RuntimeError('This event loop is already running')

有人可以向我解释为什么或如何解决这个问题吗？

非常感谢！

编辑

感谢@Dandekar，我添加了一些调试输出，参见上面的代码，并从机器人获得以下输出：

- Loop until complete...
- Starting Thread...
- Running in thread...
==============================================================================
Sample :: Just A Sample                                                       
==============================================================================
Do anything                                                           - Loop until complete...
| FAIL |
Initializing test library 'SampleProtTest' with no arguments failed: This event loop is already running
Traceback (most recent call last):
  File "C:\share\TestAutomation\SampleProtTest.py", line 42, in __init__
    self._myhost = PropHost(ip, port)
  File "C:\share\TestAutomation\SampleProtTest.py", line 24, in __init__
    _, self.__proto = self.loop.run_until_complete(self.__coro)
  File "c:\users\muechr\appdata\local\programs\python\python36\lib\asyncio\base_events.py", line 454, in run_until_complete
    self.run_forever()
  File "c:\users\muechr\appdata\local\programs\python\python36\lib\asyncio\base_events.py", line 408, in run_forever
    raise RuntimeError('This event loop is already running')
------------------------------------------------------------------------------
Sample :: Just A Sample                                               | FAIL |
1 critical test, 0 passed, 1 failed
1 test total, 0 passed, 1 failed
==============================================================================
Output:  C:\share\TestAutomation\results\output.xml
Log:     C:\share\TestAutomation\results\log.html
Report:  C:\share\TestAutomation\results\report.html

在我看来，问题是它已经在测试用例之前启动了线程。奇怪的是，如果我删除该行

_, self.__proto = self.loop.run_until_complete(self.__coro)

它似乎通过了 - 但我无法解释为什么......但这不是一个实际的解决方案，因为我无法像这样访问 __proto......

Answer 1

编辑：注释掉代码在开始时运行的部分

# if __name__=="__main__":
#    tester = SampleProtTest()
#    tester.do_something()

当您在机器人框架中导入脚本时，该部分会运行（导致端口被占用）。

另外：如果您只是尝试异步运行关键字，那么有一个库可以做到这一点（尽管我自己没有尝试过）。

robotframework-async