test_component.py

import my.component
import pytest
import asyncio


class multisend:
    def __init__(self, loop):
        self.loop = loop
        self.output_ready = loop.create_future()
        self.output = []

    def __call__(self, data):
        trigger = (len(self.output) == 0)
        self.output.append(data)
        if trigger:
            self.output_ready.set_result(True)

    async def data(self):
        if len(self.output) == 0:
            await self.output_ready
        r = self.output
        self.output_ready = self.loop.create_future()
        self.output = []
        return r


@pytest.fixture()
@pytest.mark.asyncio
def component(event_loop):
    class AdaptedComponent(my.component.Component):
        def __init__(self, loop):
            my.component.Component.__init__(self, 'test.example', 'foo', 'hostname', 1234)
            self.loop = loop
            self.send = multisend(loop)
            self.init_parser()
    yield AdaptedComponent(event_loop)


@pytest.mark.asyncio
async def test_connection_established(component):
    component.data_received('<stream:stream xmlns="jabber:component:accept" xmlns:stream="http://etherx.jabber.org/stream" id="static" from="test.example" to="test.example">')
    data = await component.send.data()
    assert(len(data) == 1)
    assert(data[0].startswith('<handshake '))
    component.data_received('<handshake xmlns="jabber:component:accept"/>')

@pytest.mark.asyncio
async def test_component_sends_stuff(component):
    data = await component.send.data()
    assert(data[0] == '<iq ... whatever the expected stuff on connect is>')


@pytest.mark.asyncio
async def test_ping(component):
    component.data_received('<iq from="foo@example.org" to="test.example" id="123" type="get"><ping xmlns="urn:xmpp:ping"/></iq>')
    data = await component.send.data()
    assert(data[0] == '<iq ... type="result"/>')