Не удается перехватить события с прослушивателем ActiveMQ на Pytest - PullRequest
Новая
       9

Не удается перехватить события с прослушивателем ActiveMQ на Pytest

0 голосов
/ 26 ноября 2018

Я пытаюсь написать контрольный пример, чтобы проверить, передается ли мое действие в тему 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 []
...