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

test: Fix test_packet_stream.py so there are no coverage issues in it

Browse Source
This commit is contained in:
Gergely Polonkai 2023-11-16 10:50:05 +01:00
parent 9e4a254127
commit a85fce7e3f
No known key found for this signature in database
GPG Key ID: 2D2885533B869ED4
1 changed files with 1 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
tests/test_packet_stream.py
View File

@ -37,7 +37,7 @@ from ssb.packet_stream import PacketStream, PSMessage, PSMessageType
async def _collect_messages(generator: AsyncIterator[Optional[PSMessage]]) -> List[Optional["PSMessage"]]: async def _collect_messages(generator: AsyncIterator[Optional[PSMessage]]) -> List[Optional["PSMessage"]]:
results = [] results = []
async for msg in generator: async for msg in generator: # pragma: no branch
results.append(msg) results.append(msg)
return results return results
@ -106,11 +106,6 @@ class MockSHSSocket(SHSDuplexStream):
yield self.output.pop(0) yield self.output.pop(0)
def disconnect(self) -> None:
"""Disconnect from the remote party"""
self.is_connected = False
class MockSHSClient(MockSHSSocket): class MockSHSClient(MockSHSSocket):
"""A mocked SHS client""" """A mocked SHS client"""
@ -120,9 +115,6 @@ class MockSHSClient(MockSHSSocket):
self.is_connected = True self.is_connected = True
for cb in self._on_connect:
await cb()
class MockSHSServer(MockSHSSocket): class MockSHSServer(MockSHSSocket):
"""A mocked SHS server""" """A mocked SHS server"""