write schedule tests

This commit is contained in:
bkfox 2023-04-02 20:31:35 +02:00
parent 6974c617a5
commit 826bb149bc
6 changed files with 141 additions and 84 deletions

View File

@ -33,7 +33,7 @@ class ScheduleAdmin(admin.ModelAdmin):
program_title.short_description = _("Program")
def freq(self, obj):
return obj.get_frequency_verbose()
return obj.get_frequency_display()
freq.short_description = _("Day")

View File

@ -1,5 +1,4 @@
import calendar
from collections import OrderedDict
import pytz
from django.db import models
@ -28,7 +27,7 @@ class Schedule(Rerun):
# Important: the first week is always the first week where the weekday of
# the schedule is present.
# For ponctual programs, there is no need for a schedule, only a diffusion
class Frequency(models.IntegerChoice):
class Frequency(models.IntegerChoices):
ponctual = 0b000000, _("ponctual")
first = 0b000001, _("1st {day} of the month")
second = 0b000010, _("2nd {day} of the month")
@ -50,7 +49,7 @@ class Schedule(Rerun):
)
timezone = models.CharField(
_("timezone"),
default=tz.get_current_timezone,
default=lambda: tz.get_current_timezone().zone,
max_length=100,
choices=[(x, x) for x in pytz.all_timezones],
help_text=_("timezone used for the date"),
@ -71,7 +70,7 @@ class Schedule(Rerun):
def __str__(self):
return "{} - {}, {}".format(
self.program.title,
self.get_frequency_verbose(),
self.get_frequency_display(),
self.time.strftime("%H:%M"),
)
@ -97,12 +96,12 @@ class Schedule(Rerun):
"""Datetime of the end."""
return self.start + utils.to_timedelta(self.duration)
def get_frequency_verbose(self):
def get_frequency_display(self):
"""Return frequency formated for display."""
from django.template.defaultfilters import date
return (
self.get_frequency_display()
self._get_FIELD_display(self._meta.get_field("frequency"))
.format(day=date(self.date, "l"))
.capitalize()
)
@ -156,16 +155,6 @@ class Schedule(Rerun):
return [self.normalize(date) for date in dates if date.month == month]
def _exclude_existing_date(self, dates):
from .diffusion import Diffusion
saved = set(
Diffusion.objects.filter(start__in=dates).values_list(
"start", flat=True
)
)
return [date for date in dates if date not in saved]
def diffusions_of_month(self, date):
"""Get episodes and diffusions for month of provided date, including
reruns.
@ -186,13 +175,11 @@ class Schedule(Rerun):
(rerun, rerun.date - self.date) for rerun in self.rerun_set.all()
]
dates = OrderedDict((date, None) for date in self.dates_of_month(date))
dates = {date: None for date in self.dates_of_month(date)}
dates.update(
[
(rerun.normalize(date.date() + delta), date)
for date in dates.keys()
for rerun, delta in reruns
]
(rerun.normalize(date.date() + delta), date)
for date in list(dates.keys())
for rerun, delta in reruns
)
# remove dates corresponding to existing diffusions

View File

@ -43,7 +43,7 @@
<section>
<h4 class="title is-4">{% translate "Diffusions" %}</h4>
{% for schedule in program.schedule_set.all %}
{{ schedule.get_frequency_verbose }}
{{ schedule.get_frequency_display }}
{% with schedule.start|date:"H:i" as start %}
{% with schedule.end|date:"H:i" as end %}
<time datetime="{{ start }}">{{ start }}</time>

View File

@ -4,47 +4,60 @@ import itertools
import pytest
from model_bakery import baker
from aircox.models import Diffusion
from aircox import models
@pytest.fixture
def stations():
return baker.make("aircox.station", quantity=2)
return baker.make("aircox.station", _quantity=2)
@pytest.fixture
def programs(stations):
return list(
items = list(
itertools.chain(
*(
baker.make("aircox.program", quantity=3, station=station)
baker.make("aircox.program", station=station, _quantity=3)
for station in stations
)
)
)
for item in items:
item.save()
return items
@pytest.fixture
def sched_initials(programs):
# use concrete class
return [
baker.make("aircox.schedule", program=program, time=time(16, 00))
# use concrete class; timezone is provided in order to ensure DST
items = [
baker.prepare(
"aircox.schedule",
program=program,
time=time(16, 00),
timezone="Europe/Brussels",
)
for program in programs
]
models.Schedule.objects.bulk_create(items)
return items
@pytest.fixture
def sched_reruns(initials):
def sched_reruns(sched_initials):
# use concrete class
return [
baker.make(
items = [
baker.prepare(
"aircox.schedule",
initial=initial,
program=initial.program,
date=initial.date,
time=(initial.start + timedelta(hours=1)).time(),
)
for initial in initials
for initial in sched_initials
]
models.Schedule.objects.bulk_create(items)
return items
@pytest.fixture
@ -54,10 +67,6 @@ def schedules(sched_initials, sched_reruns):
@pytest.fixture
def episodes(programs):
items = []
for program in programs:
items += [
baker.make("aircox.episode", parent=program, type=type)
for type, _ in Diffusion.TYPE_CHOICES
]
return items
return [
baker.make("aircox.episode", parent=program) for program in programs
]

View File

@ -10,60 +10,60 @@ from aircox.models import Schedule
class TestRerunQuerySet:
@pytest.mark.django_db
def test_station_by_obj(self, stations):
def test_station_by_obj(self, stations, schedules):
for station in stations:
queryset = (
Schedule.objects.station(station)
.distinct()
.value_list("station", flat=True)
.values_list("program__station", flat=True)
)
assert queryset.count() == 1
assert queryset.first() == station
assert queryset.first() == station.pk
@pytest.mark.django_db
def test_station_by_id(self, stations):
def test_station_by_id(self, stations, schedules):
for station in stations:
queryset = (
Schedule.objects.station(id=station.pk)
.distinct()
.value_list("station", flat=True)
.values_list("program__station", flat=True)
)
assert queryset.count() == 1
assert queryset.first() == station
assert queryset.first() == station.pk
@pytest.mark.django_db
def test_program_by_obj(self, programs):
def test_program_by_obj(self, programs, schedules):
for program in programs:
queryset = (
Schedule.objects.program(program)
.distinct()
.value_list("program", flat=True)
.values_list("program", flat=True)
)
assert queryset.count() == 1
assert queryset.first() == program
assert queryset.first() == program.pk
@pytest.mark.django_db
def test_program_by_id(self, programs):
def test_program_by_id(self, programs, schedules):
for program in programs:
queryset = (
Schedule.objects.program(id=program.pk)
.distinct()
.value_list("program", flat=True)
.values_list("program", flat=True)
)
assert queryset.count() == 1
assert queryset.first() == program
assert queryset.first() == program.pk
@pytest.mark.django_db
def test_rerun(self, sched_reruns):
queryset = Schedule.objects.rerun().value_list("initial", flat=True)
def test_rerun(self, schedules):
queryset = Schedule.objects.rerun().values_list("initial", flat=True)
assert None not in queryset
@pytest.mark.django_db
def test_initial(self, sched_initials):
def test_initial(self, schedules):
queryset = (
Schedule.objects.rerun()
Schedule.objects.initial()
.distinct()
.value_list("initial", flat=True)
.values_list("initial", flat=True)
)
assert queryset.count() == 1
assert queryset.first() is None

View File

@ -1,7 +1,13 @@
from datetime import date, datetime, time, timedelta
import pytest
from datetime import datetime
from model_bakery import baker
import calendar
from dateutil.relativedelta import relativedelta
from aircox import utils
from aircox.models import Diffusion, Schedule
class TestSchedule:
@ -32,43 +38,98 @@ class TestSchedule:
delta = utils.to_timedelta(schedule.duration)
assert schedule.end - schedule.start == delta
# def test_get_frequency_verbose(self):
# def test_get_frequency_display(self):
# pass
@pytest.mark.django_db
def test_normalize(self, schedules):
for schedule in schedules:
dt = datetime.combine(schedule.date, schedule.time)
assert schedule.normalize(dt).tzinfo.zone == schedule.timezone.zone
assert schedule.normalize(dt).tzinfo.zone == schedule.timezone
@pytest.mark.django_db
def test_dates_of_month_ponctual(self):
pass
schedule = baker.prepare(
Schedule, frequency=Schedule.Frequency.ponctual
)
at = schedule.date + relativedelta(months=4)
assert schedule.dates_of_month(at) == []
@pytest.mark.django_db
def test_dates_of_month_n_day_of_month(self):
pass
@pytest.mark.parametrize("months", range(0, 25, 2))
@pytest.mark.parametrize("hour", range(0, 24, 3))
def test_dates_of_month_last(self, months, hour):
schedule = baker.prepare(
Schedule, time=time(hour, 00), frequency=Schedule.Frequency.last
)
at = schedule.date + relativedelta(months=months)
datetimes = schedule.dates_of_month(at)
assert len(datetimes) == 1
dt = datetimes[0]
self._assert_date(schedule, at, dt)
month_info = calendar.monthrange(at.year, at.month)
at = date(at.year, at.month, month_info[1])
if at.weekday() < schedule.date.weekday():
at -= timedelta(days=7)
at += timedelta(days=schedule.date.weekday()) - timedelta(
days=at.weekday()
)
assert dt.date() == at
# since the same method is used for first, second, etc. frequencies
# we assume testing every is sufficient
@pytest.mark.django_db
@pytest.mark.parametrize("months", range(0, 25, 2))
@pytest.mark.parametrize("hour", range(0, 24, 3))
def test_dates_of_month_every(self, months, hour):
schedule = baker.prepare(
Schedule, time=time(hour, 00), frequency=Schedule.Frequency.every
)
at = schedule.date + relativedelta(months=months)
datetimes = schedule.dates_of_month(at)
last = None
for dt in datetimes:
self._assert_date(schedule, at, dt)
if last:
assert (dt - last).days == 7
last = dt
@pytest.mark.django_db
def test_dates_of_month_first_and_third(self):
pass
@pytest.mark.parametrize("months", range(0, 25, 2))
@pytest.mark.parametrize("hour", range(0, 24, 3))
def test_dates_of_month_one_on_two(self, months, hour):
schedule = baker.prepare(
Schedule,
time=time(hour, 00),
frequency=Schedule.Frequency.one_on_two,
)
at = schedule.date + relativedelta(months=months)
datetimes = schedule.dates_of_month(at)
for dt in datetimes:
self._assert_date(schedule, at, dt)
delta = dt.date() - schedule.date
assert delta.days % 14 == 0
def _assert_date(self, schedule, at, dt):
assert dt.year == at.year
assert dt.month == at.month
assert dt.weekday() == schedule.date.weekday()
assert dt.time() == schedule.time
assert dt.tzinfo.zone == schedule.timezone
@pytest.mark.django_db
def test_dates_of_month_second_and_fourth(self):
pass
def test_diffusions_of_month(self, sched_initials):
# TODO: test values of initial, rerun
for schedule in sched_initials:
at = schedule.start + timedelta(days=30)
dates = set(schedule.dates_of_month(at))
episodes, diffusions = schedule.diffusions_of_month(at)
@pytest.mark.django_db
def test_dates_of_month_every(self):
pass
@pytest.mark.django_db
def test_dates_of_month_one_on_two(self):
pass
@pytest.mark.django_db
def test__exclude_existing_date(self):
pass
@pytest.mark.django_db
def test_diffusions_of_month(self):
pass
assert all(r.date in dates for r in episodes)
assert all(
(not r.initial or r.date in dates)
and r.type == Diffusion.TYPE_ON_AIR
for r in diffusions
)