python - 无法在 Pytest 上使用 ActiveMQ 侦听器捕获事件
问题描述
我正在尝试编写一个测试用例来查看我的操作是否正在传输到 ActiveMQ 主题中。我在 pytest 之外检查我的代码,不知何故代码运行顺利,我可以看到我的消息在 ActiveMQ 主题上生成了一个事件。但是如果我在 pytest 案例中运行它,我就无法捕捉到这些事件。我似乎找不到原因。我们正在设置接收器,但如果我们使用 pytest,我们不会得到标题列表。外部 pytest 工作正常。
没有 pytest 的版本:
import messenger
import client
import time
import amq_client as aq
lst=aq.MyListener(aq.conn)
aq.conn.set_listener('', lst)
aq.conn.start()
aq.conn.connect(login=aq.user, passcode=aq.password)
aq.conn.subscribe(destination=aq.destination, ack='auto', id=1234)
print("Waiting for messages...")
resp=messenger.Messenger().send_message()
print(resp)
lognum=resp[1].split("=")[1]
time.sleep(10)
actions = [x['ENTITY_ACTION'] for x in lst.header_list]
accounts = [x['ACCOUNT'] for x in lst.header_list]
print('actions ', actions)
print(' ')
print(' ')
print(' ')
print(' ')
print('ACCOUNTS', accounts)
if(str(lognum) in accounts and 'NEW_ACCOUNT' in actions ):
print('By this it WORKS')
输出:
actions ['NEW_ACCOUNT', 'ACCOUNT_GROUP_CHANGED']
ACCOUNTS ['720308353', '720308353']
By this it WORKS
pytest版本:
import pytest
import messenger
import amq_client as aq
import time
class Test_AmqTestCase(object):
def test_newaccount(self):
lst=aq.MyListener(aq.conn)
aq.conn.set_listener('', lst)
aq.conn.start()
aq.conn.connect(login=aq.user, passcode=aq.password)
aq.conn.subscribe(destination=aq.destination, ack='auto', id=1234)
print("Waiting for messages...")
resp=messenger.Messenger().send_message()
print(resp)
lognum=resp[1].split("=")[1]
t_end=time.time()+30
t_start=time.time()
while time.time() < t_end:
time.sleep(1)
t_total=time.time()-t_start
print('Total time for event occur :', t_total)
print(' ')
actions = [x['ENTITY_ACTION'] for x in lst.header_list]
accounts = [x['ACCOUNT'] for x in lst.header_list]
print('ACCOUNTS', accounts)
assert (str(lognum) in accounts and 'NEW_ACCOUNT' in actions)
pytest结果:
assert (str(lognum) in accounts and 'NEW_ACCOUNT' in actions)
E AssertionError: assert ('720308351' in [])
E + where '720308351' = str('720308351')
Test_AmqCase.py:33: AssertionError
---------------------------- Captured stdout call -----------------------------
Waiting for messages...
['OK', 'LOGIN=720308351', 'end', '']
Total time for event occur : 30.017589569091797
ACCOUNTS []
解决方案
推荐阅读
- presto - 在 presto 中映射重复值
- sql - 从表 2 中删除表 1 中存在的行并将这些额外的行添加到表 1
- ubuntu - 无法降级我的内核版本 Ubuntu 20.04
- excel - 在 Excel VBA 中,尝试将输入的月份数转换为完整日期
- javascript - 使用外部函数的结果
- r-markdown - 我可以将用 slidy 或 ioslide 制作的幻灯片转换为 Xaringan
- php - 为什么我的计数没有随着我的 while 语句 php 增加?
- python - 我的功能 menu.displayHomeMenu(m) 陷入无限循环。为什么?
- python - 在 Python 中将 1-1000 的数字打印为单词
- android - 错误:找不到符号导入 dagger.hilt.android.components.ApplicationComponent;