gergely/pyssb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add test for on_connect

Browse Source
This commit is contained in:
Pedro Ferreira 2017-07-31 23:53:08 +02:00
parent 00fc46890e
commit 679cfb3fa1
1 changed files with 24 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
ssb/tests/test_packet_stream.py
View File

@ -1,8 +1,9 @@
import pytest import pytest
from asyncio import gather from asyncio import ensure_future, gather, Event
from asynctest import patch from asynctest import patch
from nacl.signing import SigningKey from nacl.signing import SigningKey
from secret_handshake.network import SHSSocket
from ssb.packet_stream import PSClient, PSServer, PSMessageType from ssb.packet_stream import PSClient, PSServer, PSMessageType
@ -28,8 +29,9 @@ MSG_BODY_2 = (b'{"previous":"%iQRhPyqmNLpGaO1Tpm1I22jqnUEwRwkCTDbwAGtM+lY=.sha25
b'mAkqqMwFWfP+eBIbc7DZ835er6r6h9CwAg==.sig.ed25519"}') b'mAkqqMwFWfP+eBIbc7DZ835er6r6h9CwAg==.sig.ed25519"}')
class MockSHSSocket(object): class MockSHSSocket(SHSSocket):
def __init__(self, *args, **kwargs): def __init__(self, *args, loop=None, **kwargs):
super(MockSHSSocket, self).__init__(loop)
self.input = [] self.input = []
self.output = [] self.output = []
self.is_connected = False self.is_connected = False
@ -61,7 +63,7 @@ class MockSHSSocket(object):
def _set_connected(self): def _set_connected(self):
self.is_connected = True self.is_connected = True
for cb in self._on_connect: for cb in self._on_connect:
self.event_loop.run_until_complete(cb()) ensure_future(cb(), loop=self.loop)
class MockSHSClient(MockSHSSocket): class MockSHSClient(MockSHSSocket):
@ -74,18 +76,28 @@ class MockSHSServer(MockSHSSocket):
@pytest.fixture @pytest.fixture
def ps_client(event_loop): def ps_client(event_loop):
client = PSClient('fake.local', 1000, SigningKey.generate(), b'\00' * 32, socket_class=MockSHSClient) client = PSClient('fake.local', 1000, SigningKey.generate(), b'\00' * 32, socket_class=MockSHSClient,
client.connection.event_loop = event_loop loop=event_loop)
client.connect() client.connect()
return client return client
@pytest.fixture @pytest.fixture
def ps_server(event_loop): def ps_server(event_loop):
server = PSServer('fake.local', 1000, SigningKey.generate(), socket_class=MockSHSServer) return PSServer('fake.local', 1000, SigningKey.generate(), socket_class=MockSHSServer, loop=event_loop)
server.connection.event_loop = event_loop
server.listen()
return server @pytest.mark.asyncio
async def test_on_connect(ps_server):
called = Event()
async def _on_connect():
called.set()
ps_server.on_connect(_on_connect)
ps_server.listen()
await called.wait()
assert ps_server.is_connected
@pytest.mark.asyncio @pytest.mark.asyncio
@ -202,9 +214,9 @@ async def test_message_stream(ps_client, mocker):
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_message_request(ps_server, mocker): async def test_message_request(ps_server, mocker):
mocker.patch.object(ps_server, 'register_handler', wraps=ps_server.register_handler) ps_server.listen()
assert ps_server.is_connected mocker.patch.object(ps_server, 'register_handler', wraps=ps_server.register_handler)
ps_server.send({ ps_server.send({
'name': ['whoami'], 'name': ['whoami'],