138 lines
3.0 KiB
Python
138 lines
3.0 KiB
Python
from datetime import datetime
|
|
import tzlocal
|
|
|
|
import pytest
|
|
|
|
from aircox.models import Station, Port
|
|
from aircox_streamer import controllers
|
|
from aircox_streamer.connector import Connector
|
|
|
|
|
|
local_tz = tzlocal.get_localzone()
|
|
|
|
|
|
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 = []
|
|
self.recv_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):
|
|
if isinstance(self.recv_data, list):
|
|
if len(self.recv_data):
|
|
data, self.recv_data = self.recv_data[0], self.recv_data[1:]
|
|
else:
|
|
data = ""
|
|
else:
|
|
data = self.recv_data
|
|
self.recv_data = self.recv_data[count:]
|
|
data = data[:count]
|
|
return data.encode("utf-8") if isinstance(data, str) else data
|
|
|
|
|
|
# -- models
|
|
@pytest.fixture
|
|
def station():
|
|
station = Station(name="test", path="/tmp", default=True, active=True)
|
|
station.save()
|
|
return station
|
|
|
|
|
|
@pytest.fixture
|
|
def station_ports(station):
|
|
ports = [
|
|
Port(
|
|
station=station,
|
|
direction=Port.DIRECTION_INPUT,
|
|
type=Port.TYPE_HTTP,
|
|
active=True,
|
|
),
|
|
Port(
|
|
station=station,
|
|
direction=Port.DIRECTION_OUTPUT,
|
|
type=Port.TYPE_FILE,
|
|
active=True,
|
|
),
|
|
]
|
|
for port in ports:
|
|
port.save()
|
|
return ports
|
|
|
|
|
|
# -- connectors
|
|
@pytest.fixture
|
|
def connector():
|
|
obj = Connector("/tmp/test.sock")
|
|
obj.socket_class = FakeSocket
|
|
yield obj
|
|
obj.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def fail_connector():
|
|
obj = Connector(FakeSocket.FAILING_ADDRESS)
|
|
obj.socket_class = FakeSocket
|
|
yield obj
|
|
obj.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def controller(station, connector):
|
|
connector.open()
|
|
return controllers.Streamer(station, connector)
|
|
|
|
|
|
@pytest.fixture
|
|
def socket(controller):
|
|
return controller.connector.socket
|
|
|
|
|
|
# -- metadata
|
|
@pytest.fixture
|
|
def metadata(controller):
|
|
return controllers.Metadata(controller, 1)
|
|
|
|
|
|
@pytest.fixture
|
|
def metadata_data_air_time():
|
|
return local_tz.localize(datetime(2023, 5, 1, 12, 10, 5))
|
|
|
|
|
|
@pytest.fixture
|
|
def metadata_data(metadata_data_air_time):
|
|
return {
|
|
"rid": 1,
|
|
"initial_uri": "request_uri",
|
|
"on_air": metadata_data_air_time.strftime("%Y/%m/%d %H:%M:%S"),
|
|
"status": "playing",
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def metadata_string(metadata_data):
|
|
return (
|
|
"\n".join(f"{key}={value}" for key, value in metadata_data.items())
|
|
+ "\nEND"
|
|
)
|