ฉันกำลังพยายามเรียกใช้ RobotFramework ด้วย asyncio ของ Python3.6
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 ฉันได้เพิ่มเอาต์พุต Debug บางส่วน ดูโค้ดด้านบน และรับเอาต์พุตต่อไปนี้จากหุ่นยนต์:
- 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
อย่างที่ฉันเห็น ปัญหาคือมันเริ่ม Thread ก่อนเคสทดสอบแล้ว แปลกถ้าผมเอาเส้นออก
_, self.__proto = self.loop.run_until_complete(self.__coro)
ดูเหมือนว่าจะผ่านไปได้ - แต่ฉันไม่สามารถอธิบายได้ว่าทำไม ... แต่นี่ไม่ใช่วิธีแก้ปัญหาที่ใช้งานได้จริงเนื่องจากฉันไม่สามารถเข้าถึง __proto เช่นนี้ได้ ...
