Autobahn Python RPC with Pytest

Question

So in the application i have to test RPC call ( there is a webserver made with Autobahn Python ). I wanna do integration test on those. So I am using Pytest. The way i do it is like that : 1) I have a test class like this :

@pytest.mark.incremental
class TestCreateUser:

    @pytest.fixture(scope="function", params="Json Test data")
    def create_user_data(self, request):
        return request.param

    def test_valid_user_creation(self, create_user_data):  
        ws_realm = config["websocket"]["realm"]

        runner = ApplicationRunner(
            url="%s://%s:%s/ws" % (
            config["websocket"]["protocol"], config["websocket"]["host"], config["websocket"]["port"]),
            realm=ws_realm,
            extra={"method": u"appname.create_user", "email": create_user_data["username"], create_user_data["password"]}
        )
        runner.run(MyComponent)

Then i have a component called "MyComponent" :

def __init__(self, config=None):
        super().__init__(config)

    @asyncio.coroutine
    def onJoin(self, details):
        try:
            result = yield from self.call(self.config.extra["method"], self.config.extra["email"], self.config.extra["password"])
        except Exception as e:
            print("Error: {}".format(e))
        else:
            print(result)

        self.leave()

    def onDisconnect(self):
        asyncio.get_event_loop().stop()

So i have a few problems with that, first i cannot get the result of the RPC call in the test class and second when i ran multiple times "test_valid_user_creation" ( with the fixture system ) i got an error like this :

./../.pyenv/versions/novalibrary/lib/python3.4/site-packages/autobahn/asyncio/wamp.py:167: in run
    (transport, protocol) = loop.run_until_complete(coro)
../../.pyenv/versions/3.4.3/lib/python3.4/asyncio/base_events.py:293: in run_until_complete
    self._check_closed()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <_UnixSelectorEventLoop running=False closed=True debug=False>

    def _check_closed(self):
        if self._closed:
>           raise RuntimeError('Event loop is closed')
E           RuntimeError: Event loop is closed

../../.pyenv/versions/3.4.3/lib/python3.4/asyncio/base_events.py:265: RuntimeError

So my question is How do i retrieve the value of the RPC call AND how can i not get the "Event loop is closed " when the test is called multiple time with the fixture.

Thank you.

Have you checked that your `def onDisconnect(self):` is called in every run? You have to close the `loop` and wait until terminated, before start a new one. About your `result`, make it a `member` of `class MyComponent` or copy to a `global var`. — stovfl, May 11 '17 at 07:53
onDisconnect() is called every time; but how do i wait to close the loop. Because when i call asyncio.get_event_loop().close() i got the exception. When i call asyncio.get_event_loop().stop() and then called asyncio.get_event_loop().is_running() it return true. — Jbeat, May 11 '17 at 19:21
My bad, I'm wrong. I meant `loop.stop()` then **wait** then `loop.close()`. Read this Answer about **wait**: http://stackoverflow.com/a/43810272/7414759 — stovfl, May 11 '17 at 20:20

Jbeat · Accepted Answer · 2017-05-12T20:09:46.623

So basically, to return the value i create another module "shared_result" and i put one global variable in there to return the value from the component into the test class. As for the "Event loop is closed" exception i was getting, i do that before each time i call the runner :

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

This is where i find the solution : https://github.com/aio-libs/aiohttp/issues/441

If somebody can comment on this and explain why this is needed and gave me a better solution, it would be great.

Autobahn Python RPC with Pytest

1 Answers1