Skip to content

Commit

Permalink
fix: make python tests mimic the rust tests
Browse files Browse the repository at this point in the history
  • Loading branch information
@cmgriffing
cmgriffing committed Feb 23, 2024
1 parent a6b9f87 commit 5baaac4
Show file tree
Hide file tree
Showing 2 changed files with 256 additions and 50 deletions.
265 changes: 239 additions & 26 deletions clients/python/tests/test_scheduler.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -28,13 +28,103 @@ def test_schedule_active_none(self):

self.assertTrue(schedule_active)

def test_schedule_active_start_end(self):
def test_schedule_active_start_end_single_day(self):
"""
Test that a ScheduleType.START_END returns accurate results
"""
now = arrow.utcnow()
zero_day = Arrow.utcfromtimestamp(0)

for hour in range(24):
mocked_now = Arrow(
now.year,
now.month,
now.day,
hour,
0,
0,
0,
dateutil_tz.tzutc(),
fold=getattr(now, "fold", 0),
)

zero_day_with_mocked_time = Arrow(
zero_day.year,
zero_day.month,
zero_day.day,
hour,
0,
0,
0,
dateutil_tz.tzutc(),
fold=getattr(zero_day, "fold", 0),
)

before_schedule = Schedule(
start=int(mocked_now.timestamp()) * 1000,
end=int(mocked_now.timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.START_END,
startTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+3).timestamp()) * 1000,
)

before_schedule_active = Scheduler.is_schedule_active_with_now(
before_schedule, ScheduleType.GLOBAL, int(mocked_now.timestamp()) * 1000
)

self.assertFalse(before_schedule_active)

during_schedule = Schedule(
start=int(mocked_now.timestamp()) * 1000,
end=int(mocked_now.timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.START_END,
startTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
)

during_schedule_active = Scheduler.is_schedule_active_with_now(
during_schedule, ScheduleType.GLOBAL, int(mocked_now.timestamp()) * 1000
)

self.assertTrue(during_schedule_active)

after_schedule = Schedule(
start=int(mocked_now.timestamp()) * 1000,
end=int(mocked_now.timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.START_END,
startTime=int(zero_day_with_mocked_time.shift(hours=-3).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
)

after_schedule_active = Scheduler.is_schedule_active_with_now(
after_schedule, ScheduleType.GLOBAL, int(mocked_now.timestamp()) * 1000
)

self.assertFalse(after_schedule_active)


def test_schedule_active_start_end_multi_day(self):
"""
Test that a ScheduleType.START_END returns accurate results
"""
now = arrow.utcnow()
zero_day = Arrow.utcfromtimestamp(0)

for hour in range(24):
zero_day_with_mocked_time = Arrow(
zero_day.year,
zero_day.month,
zero_day.day,
hour,
0,
0,
0,
dateutil_tz.tzutc(),
fold=getattr(zero_day, "fold", 0),
)

mocked_now = Arrow(
now.year,
Expand All @@ -49,16 +139,16 @@ def test_schedule_active_start_end(self):
)

before_schedule = Schedule(
start=int(mocked_now.shift(days=1).timestamp()) * 1000,
start=int(mocked_now.timestamp()) * 1000,
end=int(mocked_now.shift(days=2).timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.START_END,
startTime=0,
endTime=0,
startTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+3).timestamp()) * 1000,
)

before_schedule_active = Scheduler.is_schedule_active(
before_schedule, ScheduleType.GLOBAL
before_schedule_active = Scheduler.is_schedule_active_with_now(
before_schedule, ScheduleType.GLOBAL, int(mocked_now.timestamp()) * 1000
)

self.assertFalse(before_schedule_active)
Expand All @@ -68,36 +158,39 @@ def test_schedule_active_start_end(self):
end=int(mocked_now.shift(days=1).timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.START_END,
startTime=0,
endTime=0,
startTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
)

during_schedule_active = Scheduler.is_schedule_active(
during_schedule, ScheduleType.GLOBAL
during_schedule_active = Scheduler.is_schedule_active_with_now(
during_schedule, ScheduleType.GLOBAL, int(mocked_now.timestamp()) * 1000
)

self.assertTrue(during_schedule_active)

after_schedule = Schedule(
start=int(mocked_now.shift(days=-2).timestamp()) * 1000,
end=int(mocked_now.shift(days=-1).timestamp()) * 1000,
end=int(mocked_now.timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.START_END,
startTime=0,
endTime=0,
startTime=int(zero_day_with_mocked_time.shift(hours=-3).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
)

after_schedule_active = Scheduler.is_schedule_active(
after_schedule, ScheduleType.GLOBAL
after_schedule_active = Scheduler.is_schedule_active_with_now(
after_schedule, ScheduleType.GLOBAL, int(mocked_now.timestamp()) * 1000
)

self.assertFalse(after_schedule_active)

def test_schedule_active_daily(self):
#

def test_schedule_active_daily_single_day(self):
"""
Test that a ScheduleType.DAILY returns accurate results
"""
now = arrow.utcnow()
zero_day = Arrow.utcfromtimestamp(0)

for hour in range(24):
mocked_now = Arrow(
Expand All @@ -111,16 +204,121 @@ def test_schedule_active_daily(self):
dateutil_tz.tzutc(),
fold=getattr(now, "fold", 0),
)
mocked_now_timestamp = mocked_now.timestamp() * 1000

zero_day_with_mocked_time = Arrow(
zero_day.year,
zero_day.month,
zero_day.day,
hour,
0,
0,
0,
dateutil_tz.tzutc(),
fold=getattr(zero_day, "fold", 0),
)

before_day_schedule = Schedule(
start=int(mocked_now.timestamp()) * 1000,
end=int(mocked_now.timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+3).timestamp()) * 1000,
)

before_day_schedule_active = Scheduler.is_schedule_active_with_now(
before_day_schedule, ScheduleType.GLOBAL, mocked_now_timestamp
)

self.assertFalse(before_day_schedule_active)

during_schedule = Schedule(
start=int(mocked_now.timestamp()) * 1000,
end=int(mocked_now.timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
)

during_schedule_active = Scheduler.is_schedule_active_with_now(
during_schedule, ScheduleType.GLOBAL, mocked_now_timestamp
)

self.assertTrue(during_schedule_active, f"Hour: {hour}")

after_day_schedule = Schedule(
start=int(mocked_now.timestamp()) * 1000,
end=int(mocked_now.timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(zero_day_with_mocked_time.shift(hours=-3).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
)

after_day_schedule_active = Scheduler.is_schedule_active_with_now(
after_day_schedule, ScheduleType.GLOBAL, mocked_now_timestamp
)

self.assertFalse(after_day_schedule_active)


def test_schedule_active_daily_multi_day(self):
"""
Test that a ScheduleType.DAILY returns accurate results
"""
now = arrow.utcnow()
zero_day = Arrow.utcfromtimestamp(0)

for hour in range(24):
mocked_now = Arrow(
now.year,
now.month,
now.day,
hour,
0,
0,
0,
dateutil_tz.tzutc(),
fold=getattr(now, "fold", 0),
)
mocked_now_timestamp = mocked_now.timestamp() * 1000

zero_day_with_mocked_time = Arrow(
zero_day.year,
zero_day.month,
zero_day.day,
hour,
0,
0,
0,
dateutil_tz.tzutc(),
fold=getattr(zero_day, "fold", 0),
)

before_whole_schedule = Schedule(
start=int(mocked_now.shift(days=+1).timestamp()) * 1000,
end=int(mocked_now.shift(days=+3).timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+3).timestamp()) * 1000,
)

before_whole_schedule_active = Scheduler.is_schedule_active_with_now(
before_whole_schedule, ScheduleType.GLOBAL, mocked_now_timestamp
)

self.assertFalse(before_whole_schedule_active)

before_day_schedule = Schedule(
start=int(mocked_now.shift(days=-1).timestamp()) * 1000,
end=int(mocked_now.shift(days=1).timestamp()) * 1000,
end=int(mocked_now.shift(days=+1).timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(mocked_now.shift(hours=1).timestamp()) * 1000,
endTime=int(mocked_now.shift(hours=3).timestamp()) * 1000,
startTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+3).timestamp()) * 1000,
)

before_day_schedule_active = Scheduler.is_schedule_active_with_now(
Expand All @@ -131,11 +329,11 @@ def test_schedule_active_daily(self):

during_schedule = Schedule(
start=int(mocked_now.shift(days=-1).timestamp()) * 1000,
end=int(mocked_now.shift(days=1).timestamp()) * 1000,
end=int(mocked_now.shift(days=+1).timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(mocked_now.shift(hours=-1).timestamp()) * 1000,
endTime=int(mocked_now.shift(hours=1).timestamp()) * 1000,
startTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=+1).timestamp()) * 1000,
)

during_schedule_active = Scheduler.is_schedule_active_with_now(
Expand All @@ -145,12 +343,12 @@ def test_schedule_active_daily(self):
self.assertTrue(during_schedule_active, f"Hour: {hour}")

after_day_schedule = Schedule(
start=int(now.shift(days=-1).timestamp()) * 1000,
end=int(now.shift(days=1).timestamp()) * 1000,
start=int(mocked_now.shift(days=-1).timestamp()) * 1000,
end=int(mocked_now.shift(days=+1).timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(mocked_now.shift(hours=-3).timestamp()) * 1000,
endTime=int(mocked_now.shift(hours=-1).timestamp()) * 1000,
startTime=int(zero_day_with_mocked_time.shift(hours=-3).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
)

after_day_schedule_active = Scheduler.is_schedule_active_with_now(
Expand All @@ -159,6 +357,21 @@ def test_schedule_active_daily(self):

self.assertFalse(after_day_schedule_active)

after_whole_schedule = Schedule(
start=int(mocked_now.shift(days=-3).timestamp()) * 1000,
end=int(mocked_now.shift(days=-1).timestamp()) * 1000,
timezone="UTC",
timeType=ScheduleTimeType.DAILY,
startTime=int(zero_day_with_mocked_time.shift(hours=-3).timestamp()) * 1000,
endTime=int(zero_day_with_mocked_time.shift(hours=-1).timestamp()) * 1000,
)

after_whole_schedule_active = Scheduler.is_schedule_active_with_now(
after_whole_schedule, ScheduleType.GLOBAL, mocked_now_timestamp
)

self.assertFalse(after_whole_schedule_active)


if __name__ == "__main__":
unittest.main()
Loading

0 comments on commit 5baaac4

Please sign in to comment.