forked from rc/aircox
rewrite tests + fix error in schedule generator
This commit is contained in:
bkfox
@ -1,4 +1,5 @@
|
||||
import datetime
|
||||
import calendar
|
||||
import os
|
||||
import shutil
|
||||
import logging
|
||||
@ -400,6 +401,18 @@ class Schedule(models.Model):
|
||||
date = date_or_default(date, True).replace(day=1)
|
||||
freq = self.frequency
|
||||
|
||||
# last of the month
|
||||
if freq == Schedule.Frequency.last:
|
||||
date = date.replace(day=calendar.monthrange(date.year, date.month)[1])
|
||||
|
||||
# end of month before the wanted weekday: move one week back
|
||||
if date.weekday() < self.date.weekday():
|
||||
date -= datetime.timedelta(days = 7)
|
||||
|
||||
delta = self.date.weekday() - date.weekday()
|
||||
date += datetime.timedelta(days = delta)
|
||||
return [self.normalize(date)]
|
||||
|
||||
# move to the first day of the month that matches the schedule's weekday
|
||||
# check on SO#3284452 for the formula
|
||||
first_weekday = date.weekday()
|
||||
@ -408,19 +421,9 @@ class Schedule(models.Model):
|
||||
- first_weekday + sched_weekday)
|
||||
month = date.month
|
||||
|
||||
# last of the month
|
||||
if freq == Schedule.Frequency.last:
|
||||
date += tz.timedelta(days = 4 * 7)
|
||||
next_date = date + tz.timedelta(days = 7)
|
||||
if next_date.month == month:
|
||||
date = next_date
|
||||
return [self.normalize(date)]
|
||||
|
||||
dates = []
|
||||
if freq == Schedule.Frequency.one_on_two:
|
||||
# NOTE previous algorithm was based on the week number, but this
|
||||
# approach is wrong because number of weeks in a year can be
|
||||
# 52 or 53. This also clashes with the first week of the year.
|
||||
# check date base on a diff of dates base on a 14 days delta
|
||||
diff = as_date(date, False) - as_date(self.date, False)
|
||||
if diff.days % 14:
|
||||
date += tz.timedelta(days = 7)
|
||||
@ -445,36 +448,31 @@ class Schedule(models.Model):
|
||||
If exclude_saved, exclude all diffusions that are yet in the database.
|
||||
"""
|
||||
dates = self.dates_of_month(date)
|
||||
saved = Diffusion.objects.filter(start__in = dates,
|
||||
program = self.program)
|
||||
diffusions = []
|
||||
|
||||
duration = utils.to_timedelta(self.duration)
|
||||
|
||||
# existing diffusions
|
||||
for item in saved:
|
||||
for item in Diffusion.objects.filter(
|
||||
program = self.program, start__in = dates):
|
||||
if item.start in dates:
|
||||
dates.remove(item.start)
|
||||
if not exclude_saved:
|
||||
diffusions.append(item)
|
||||
|
||||
# others
|
||||
for date in dates:
|
||||
first_date = date
|
||||
if self.initial:
|
||||
first_date -= self.date - self.initial.date
|
||||
|
||||
first_diffusion = Diffusion.objects.filter(start = first_date,
|
||||
program = self.program)
|
||||
first_diffusion = first_diffusion[0] if first_diffusion.count() \
|
||||
else None
|
||||
diffusions.append(Diffusion(
|
||||
program = self.program,
|
||||
type = Diffusion.Type.unconfirmed,
|
||||
initial = first_diffusion if self.initial else None,
|
||||
start = date,
|
||||
end = date + duration,
|
||||
))
|
||||
# new diffusions
|
||||
duration = utils.to_timedelta(self.duration)
|
||||
if self.initial:
|
||||
delta = self.date - self.initial.date
|
||||
diffusions += [
|
||||
Diffusion(
|
||||
program = self.program,
|
||||
type = Diffusion.Type.unconfirmed,
|
||||
initial = \
|
||||
Diffusion.objects.filter(start = date - delta).first() \
|
||||
if self.initial else None,
|
||||
start = date,
|
||||
end = date + duration,
|
||||
) for date in dates
|
||||
]
|
||||
return diffusions
|
||||
|
||||
def __str__(self):
|
||||
|
||||
@ -1,79 +1,66 @@
|
||||
import datetime
|
||||
import calendar
|
||||
import logging
|
||||
from dateutil.relativedelta import relativedelta
|
||||
|
||||
from django.test import TestCase
|
||||
from django.utils import timezone as tz
|
||||
|
||||
from aircox.programs.models import *
|
||||
|
||||
logger = logging.getLogger('aircox.test')
|
||||
logger.setLevel('INFO')
|
||||
|
||||
class Programs (TestCase):
|
||||
def setUp (self):
|
||||
stream = Stream.objects.get_or_create(
|
||||
name = 'diffusions',
|
||||
defaults = { 'type': Stream.Type.schedule }
|
||||
)[0]
|
||||
Program.objects.create(name = 'source', stream = stream)
|
||||
Program.objects.create(name = 'microouvert', stream = stream)
|
||||
class ScheduleCheck (TestCase):
|
||||
def setUp(self):
|
||||
self.schedules = [
|
||||
Schedule(
|
||||
date = tz.now(),
|
||||
duration = datetime.time(1,30),
|
||||
frequency = frequency,
|
||||
)
|
||||
for frequency in Schedule.Frequency.__members__.values()
|
||||
]
|
||||
|
||||
self.schedules = {}
|
||||
self.programs = {}
|
||||
def test_frequencies(self):
|
||||
for schedule in self.schedules:
|
||||
logger.info('- test frequency %s' % schedule.get_frequency_display())
|
||||
date = schedule.date
|
||||
count = 24
|
||||
while count:
|
||||
logger.info('- month %(month)s/%(year)s' % {
|
||||
'month': date.month,
|
||||
'year': date.year
|
||||
})
|
||||
count -= 1
|
||||
dates = schedule.dates_of_month(date)
|
||||
if schedule.frequency == schedule.Frequency.one_on_two:
|
||||
self.check_one_on_two(schedule, date, dates)
|
||||
elif schedule.frequency == schedule.Frequency.last:
|
||||
self.check_last(schedule, date, dates)
|
||||
else:
|
||||
pass
|
||||
date += relativedelta(months = 1)
|
||||
|
||||
def test_create_programs_schedules (self):
|
||||
program = Program.objects.get(name = 'source')
|
||||
def check_one_on_two(self, schedule, date, dates):
|
||||
for date in dates:
|
||||
delta = date.date() - schedule.date.date()
|
||||
self.assertEqual(delta.days % 14, 0)
|
||||
|
||||
sched_0 = self.create_schedule(program, 'one on two', [
|
||||
tz.datetime(2015, 10, 2, 18),
|
||||
tz.datetime(2015, 10, 16, 18),
|
||||
tz.datetime(2015, 10, 30, 18),
|
||||
]
|
||||
)
|
||||
sched_1 = self.create_schedule(program, 'one on two', [
|
||||
tz.datetime(2015, 10, 5, 18),
|
||||
tz.datetime(2015, 10, 19, 18),
|
||||
],
|
||||
rerun = sched_0
|
||||
)
|
||||
def check_last(self, schedule, date, dates):
|
||||
month_info = calendar.monthrange(date.year, date.month)
|
||||
date = datetime.date(date.year, date.month, month_info[1])
|
||||
|
||||
self.programs[program.pk] = program
|
||||
# end of month before the wanted weekday: move one week back
|
||||
if date.weekday() < schedule.date.weekday():
|
||||
date -= datetime.timedelta(days = 7)
|
||||
|
||||
program = Program.objects.get(name = 'microouvert')
|
||||
# special case with november first week starting on sunday
|
||||
sched_2 = self.create_schedule(program, 'first and third', [
|
||||
tz.datetime(2015, 11, 6, 18),
|
||||
tz.datetime(2015, 11, 20, 18),
|
||||
],
|
||||
date = tz.datetime(2015, 10, 23, 18),
|
||||
)
|
||||
date -= datetime.timedelta(days = date.weekday())
|
||||
date += datetime.timedelta(days = schedule.date.weekday())
|
||||
self.assertEqual(date, dates[0].date())
|
||||
|
||||
def check_n_of_week(self, schedule, date, dates):
|
||||
pass
|
||||
|
||||
def create_schedule (self, program, frequency, dates, date = None, rerun = None):
|
||||
frequency = Schedule.Frequency[frequency]
|
||||
schedule = Schedule(
|
||||
program = program,
|
||||
frequency = frequency,
|
||||
date = date or dates[0],
|
||||
rerun = rerun,
|
||||
duration = datetime.time(1, 30)
|
||||
)
|
||||
print(schedule.__dict__)
|
||||
schedule.save()
|
||||
|
||||
self.schedules[schedule.pk] = (schedule, dates)
|
||||
return schedule
|
||||
|
||||
def test_check_schedule (self):
|
||||
for schedule, dates in self.schedules:
|
||||
dates = [ tz.make_aware(date) for date in dates ]
|
||||
dates.sort()
|
||||
|
||||
# dates
|
||||
dates_ = schedule.dates_of_month(dates[0])
|
||||
dates_.sort()
|
||||
self.assertEqual(dates_, dates)
|
||||
|
||||
# diffusions
|
||||
dates_ = schedule.diffusions_of_month(dates[0])
|
||||
dates_ = [date_.date for date_ in dates_]
|
||||
dates_.sort()
|
||||
self.assertEqual(dates_, dates)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user