183 lines
4.1 KiB
Python
183 lines
4.1 KiB
Python
import os
|
|
|
|
from datetime import datetime, time
|
|
import tzlocal
|
|
|
|
import pytest
|
|
|
|
from aircox import models
|
|
from aircox_streamer import controllers
|
|
from aircox_streamer.connector import Connector
|
|
|
|
|
|
local_tz = tzlocal.get_localzone()
|
|
|
|
|
|
working_dir = os.path.join(os.path.dirname(__file__), "working_dir")
|
|
|
|
|
|
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.decode())
|
|
|
|
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
|
|
) or b"\nEND"
|
|
|
|
def is_sent(self, data):
|
|
"""Return True if provided data have been sent."""
|
|
# use [:-1] because connector add "\n" at sent data
|
|
return any(r for r in self.sent_data if r == data or r[:-1] == data)
|
|
|
|
|
|
# -- models
|
|
@pytest.fixture
|
|
def station():
|
|
station = models.Station(
|
|
name="test", path=working_dir, default=True, active=True
|
|
)
|
|
station.save()
|
|
return station
|
|
|
|
|
|
@pytest.fixture
|
|
def station_ports(station):
|
|
items = [
|
|
models.Port(
|
|
station=station,
|
|
direction=models.Port.DIRECTION_INPUT,
|
|
type=models.Port.TYPE_HTTP,
|
|
active=True,
|
|
),
|
|
models.Port(
|
|
station=station,
|
|
direction=models.Port.DIRECTION_OUTPUT,
|
|
type=models.Port.TYPE_FILE,
|
|
active=True,
|
|
),
|
|
]
|
|
models.Port.objects.bulk_create(items)
|
|
return items
|
|
|
|
|
|
@pytest.fixture
|
|
def program(station):
|
|
program = models.Program(title="test", station=station)
|
|
program.save()
|
|
return program
|
|
|
|
|
|
@pytest.fixture
|
|
def stream(program):
|
|
stream = models.Stream(
|
|
program=program, begin=time(10, 12), end=time(12, 13)
|
|
)
|
|
stream.save()
|
|
return stream
|
|
|
|
|
|
@pytest.fixture
|
|
def sounds(program):
|
|
items = [
|
|
models.Sound(
|
|
name=f"sound {i}",
|
|
program=program,
|
|
type=models.Sound.TYPE_ARCHIVE,
|
|
position=i,
|
|
file=f"sound-{i}.mp3",
|
|
)
|
|
for i in range(0, 3)
|
|
]
|
|
models.Sound.objects.bulk_create(items)
|
|
return items
|
|
|
|
|
|
# -- connectors
|
|
@pytest.fixture
|
|
def connector():
|
|
obj = Connector(os.path.join(working_dir, "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"
|
|
)
|