forked from rc/aircox
57 lines
1.6 KiB
Python
57 lines
1.6 KiB
Python
from datetime import date, datetime, timedelta, time
|
|
from django.utils import timezone as tz
|
|
|
|
import pytest
|
|
|
|
from aircox import models
|
|
from aircox.controllers import diffusion_monitor
|
|
|
|
now = date.today()
|
|
|
|
|
|
@pytest.fixture
|
|
def monitor(logger):
|
|
logger = logger._imeta.clone().inject(diffusion_monitor, "logger")
|
|
yield diffusion_monitor.DiffusionMonitor(date=now)
|
|
logger.release()
|
|
|
|
|
|
class TestDiffusion:
|
|
@pytest.mark.django_db
|
|
def test_update(self, monitor, schedules, sched_initials, logger):
|
|
monitor.update()
|
|
|
|
diffusions = models.Diffusion.objects.filter(
|
|
schedule__in=sched_initials
|
|
)
|
|
by_date = {}
|
|
for diff in diffusions:
|
|
assert diff.episode_id
|
|
by_date.setdefault(diff.schedule_id, set()).add(
|
|
(diff.start, diff.end)
|
|
)
|
|
|
|
for schedule in sched_initials:
|
|
if schedule.pk not in by_date:
|
|
continue
|
|
_, items = schedule.diffusions_of_month(now)
|
|
assert all(
|
|
(item.start, item.end) in by_date[schedule.pk]
|
|
for item in items
|
|
)
|
|
|
|
@pytest.mark.django_db
|
|
def test_clean(self, monitor, episode):
|
|
start = tz.make_aware(
|
|
datetime.combine(monitor.date - timedelta(days=1), time(10, 20))
|
|
)
|
|
diff = models.Diffusion(
|
|
type=models.Diffusion.TYPE_UNCONFIRMED,
|
|
episode=episode,
|
|
start=start,
|
|
end=start + timedelta(minutes=30),
|
|
)
|
|
diff.save()
|
|
monitor.clean()
|
|
assert not models.Diffusion.objects.filter(pk=diff.pk).first()
|