rc/aircox
13
1
Fork 1
Code Issues 15 Pull Requests Projects Releases Wiki Activity

handle_diffusion tests

Browse Source
This commit is contained in:
bkfox 2023-06-18 18:50:08 +02:00
parent b453c821c7
commit 05c6cbc839
3 changed files with 96 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
aircox_streamer/controllers/monitor.py
View File

@ -223,9 +223,8 @@ class Monitor:
or dealer.remaining < self.delay.total_seconds()
):
self.start_diff(dealer, diff)
# cancel
if diff.start < now - self.cancel_timeout:
elif diff.start < now - self.cancel_timeout:
self.cancel_diff(dealer, diff)
def log(self, source, **kwargs):

3
aircox_streamer/tests/conftest.py
View File

@ -318,7 +318,8 @@ def streamer(station, station_ports):
streamer.sources = [
FakePlaylist(i, uri=f"source-{i}") for i in range(0, 3)
]
streamer.sources.append(FakeQueueSource(len(streamer.sources)))
streamer.dealer = FakeQueueSource(len(streamer.sources))
streamer.sources.append(streamer.dealer)
return streamer

100
aircox_streamer/tests/test_controllers_monitor.py
View File

@ -20,13 +20,13 @@ def monitor(streamer):
@pytest.fixture
def diffusion(program, episode):
def diffusion(program, episode, sound):
return baker.make(
models.Diffusion,
program=program,
episode=episode,
start=tz.now() - tz.timedelta(minutes=10),
end=tz.now() + tz.timedelta(minutes=30),
start=tz.now() - tz.timedelta(minutes=2),
end=tz.now() + tz.timedelta(minutes=4),
schedule=None,
type=models.Diffusion.TYPE_ON_AIR,
)
@ -64,8 +64,10 @@ def log(station, source, sound):
class TestMonitor:
@pytest.mark.django_db(transaction=True)
def test_last_diff_start(self, monitor):
pass
def test_last_diff_start(self, monitor, diffusion, log):
log.diffusion = diffusion
log.save()
assert monitor.last_diff_start == log
@pytest.mark.django_db(transaction=True)
def test___init__(self, monitor):
@ -188,8 +190,92 @@ class TestMonitor:
monitor.trace_tracks(log)
@pytest.mark.django_db(transaction=True)
def test_handle_diffusions(self, monitor):
pass
def test_handle_diffusions(self, monitor, streamer, diffusion, sound):
interface(
monitor,
{
"start_diff": None,
"cancel_diff": None,
},
)
streamer.dealer.queue = [1]
streamer.dealer.rid = None
streamer.dealer.remaining = monitor.delay.total_seconds() - 10
monitor.handle_diffusions()
assert monitor.calls["start_diff"] == (
(streamer.dealer, diffusion),
{},
)
assert not monitor.calls["cancel_diff"]
@pytest.mark.django_db(transaction=True)
def test_handle_diffusions_returns_on_diff(
self, monitor, streamer, diffusion, log
):
interface(
monitor,
{
"start_diff": None,
"cancel_diff": None,
},
)
streamer.dealer.queue = [1]
streamer.dealer.rid = None
streamer.dealer.remaining = monitor.delay.total_seconds() - 10
diffusion.start = tz.now() - tz.timedelta(minutes=30)
diffusion.end = tz.now() - tz.timedelta(minutes=20)
diffusion.save()
monitor.handle_diffusions()
assert not monitor.calls["start_diff"]
assert not monitor.calls["cancel_diff"]
@pytest.mark.django_db(transaction=True)
def test_handle_diffusions_returns_on_diff_log_exists(
self, monitor, streamer, diffusion, log
):
interface(
monitor,
{
"start_diff": None,
"cancel_diff": None,
},
)
streamer.dealer.queue = [1]
streamer.dealer.rid = None
streamer.dealer.remaining = monitor.delay.total_seconds() - 10
log.diffusion = diffusion
log.save()
monitor.handle_diffusions()
assert not monitor.calls["start_diff"]
assert not monitor.calls["cancel_diff"]
@pytest.mark.django_db(transaction=True)
def test_handle_diffusions_cancel_diff(self, monitor, streamer, diffusion):
interface(
monitor,
{
"start_diff": None,
"cancel_diff": None,
},
)
streamer.dealer.queue = None
streamer.dealer.rid = "13"
streamer.dealer.remaining = monitor.delay.total_seconds() + 10
diffusion.start = (
tz.now() - monitor.cancel_timeout - tz.timedelta(seconds=30)
)
diffusion.end = tz.now() + tz.timedelta(minutes=30)
diffusion.save()
monitor.handle_diffusions()
assert not monitor.calls["start_diff"]
assert monitor.calls["cancel_diff"] == (
(streamer.dealer, diffusion),
{},
)
@pytest.mark.django_db(transaction=True)
def test_log(self, monitor, source):