Я пытаюсь запустить RobotFramework с помощью asyncio Python3.6.
Соответствующий код Python выглядит следующим образом:
""" 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 вот так...