52 lines
1.1 KiB
Python
52 lines
1.1 KiB
Python
import pytest
|
|
|
|
from aircox_streamer import connector
|
|
|
|
|
|
class FakeSocket:
|
|
FAILING_ADDRESS = -1
|
|
"""Connect with this address fails."""
|
|
|
|
family, type, address = None, None, None
|
|
sent_data = None
|
|
"""List of data that have been `send[all]`"""
|
|
recv_data = None
|
|
"""Response data to return on recv."""
|
|
|
|
def __init__(self, family, type):
|
|
self.family = family
|
|
self.type = type
|
|
self.sent_data = []
|
|
|
|
def connect(self, address):
|
|
if address == self.FAILING_ADDRESS:
|
|
raise RuntimeError("invalid connection")
|
|
self.address = address
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
def sendall(self, data):
|
|
self.sent_data.append(data)
|
|
|
|
def recv(self, count):
|
|
data = self.recv_data[:count]
|
|
self.recv_data = data[count:]
|
|
return data.encode("utf-8") if isinstance(data, str) else data
|
|
|
|
|
|
class Connector(connector.Connector):
|
|
socket_class = FakeSocket
|
|
|
|
|
|
@pytest.fixture
|
|
def connector(request):
|
|
obj = Connector("test")
|
|
yield obj
|
|
obj.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def fail_connector():
|
|
return Connector(FakeSocket.FAILING_ADDRESS)
|