Btukfyl

I'm trying to write a test case to see if my action is being transmitted into ActiveMQ topic.
I check my code outside of pytest, somehow code runs smoothly and I can see my message generated an event on ActiveMQ topic. But If I run it inside pytest case, I can't catch the events. I can't seem to find a reason. We are setting the receiver but we don't get header list if we use pytest. Outside pytest works fine.

Version without 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')

Output:

actions ['NEW_ACCOUNT', 'ACCOUNT_GROUP_CHANGED']









ACCOUNTS ['720308353', '720308353']

By this it WORKS

Pytest version:

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 result:

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