import pytest from rota.shifts import NoWorkers, RotaBuilder, SingleShift, WarningTermination, WorkerRequirement, days import datetime from rota.workers import NonWorkingDays, NotAvailableToWork, Worker, generate_not_available_to_works from rota.workers import WorkRequests def generate_basic_rota( weeks_to_rota=10, workers=0, start_date=datetime.date(2022, 3, 7) ): Rota = RotaBuilder( start_date, weeks_to_rota=weeks_to_rota, ) # Add a few workers if workers: Rota.add_workers( [ Worker(name=f"worker{i:02}", site="group1", grade=1) for i in range(1, workers + 1) ] ) return Rota class TestWorkers: def test_no_workers(self): """Check error raised if attempt to solve with no workers""" Rota = generate_basic_rota() Rota.add_shifts( SingleShift( sites=("group1",), name="d", length=12.5, days=days[:5], balance_offset=10, workers_required=1, ), SingleShift( sites=("group1",), name="w", length=12.5, days=days[5:], workers_required=1, ), ) with pytest.raises(NoWorkers): Rota.build_and_solve(solve=False) def test_workers_duplicate_name(self): Rota = generate_basic_rota() workers = [] workers.append(Worker(name="worker1", site="group1", grade=1, fte=100)) workers.append(Worker(name="worker1", site="group1", grade=1, fte=100)) workers.append(Worker(name="worker3", site="group2", grade=1, fte=50)) workers.append(Worker(name="worker4", site="group2", grade=1, fte=50)) Rota.add_workers(workers) Rota.add_shifts( SingleShift( sites=("group1",), name="d", length=12.5, days=days[:5], balance_offset=10, workers_required=1, ), ) with pytest.raises(WarningTermination): Rota.build_and_solve(solve=False) def test_worker_ftes(self): weeks_to_rota = 10 start_date = datetime.date(2022, 3, 7) Rota = RotaBuilder( start_date, weeks_to_rota=weeks_to_rota, ) worker1_end_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 2) worker2_start_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 2) worker3_start_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 4) worker3_end_date = start_date + datetime.timedelta(weeks=3 * weeks_to_rota / 4) workers = [] workers.append( Worker( name="worker1", site="group1", grade=1, fte=100, end_date=worker1_end_date, ) ) workers.append( Worker( name="worker2", site="group1", grade=1, fte=100, start_date=worker2_start_date, ) ) workers.append( Worker( name="worker3", site="group2", grade=1, fte=100, start_date=worker3_start_date, end_date=worker3_end_date, ) ) workers.append(Worker(name="worker4", site="group2", grade=1, fte=50)) workers.append( Worker(name="worker5", site="group2", grade=1, fte=50, nwd=["Mon", "Tue"]) ) workers.append( Worker( name="worker6", site="group3", grade=1, fte=100, start_date=start_date + datetime.timedelta(weeks=weeks_to_rota), ) ) workers.append( Worker(name="worker7", site="group3", grade=1, fte=50, end_date=start_date) ) Rota.add_workers(workers) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="d", length=12.5, days=days[:5], balance_offset=10, workers_required=1, ), ) with pytest.raises(WarningTermination): Rota.build_and_solve() Rota.terminate_on_warning.remove("Worker/no valid shifts") Rota.build_and_solve() # All workers should have an adjusted fte of 50 % for worker in Rota.get_workers(): # Except those in group3 if worker.site == "group3": assert worker.fte_adj == 0 assert Rota.get_worker_shift_list(worker).count("d") == 0 continue assert worker.fte_adj == 50 if worker.fte == 100: assert worker.proportion_rota_to_work == 0.5 assert worker.days_to_work == 35 assert Rota.get_worker_shift_list(worker).count("d") == 10 if worker.name == "worker1": assert Rota.get_worker_shift_list_string(worker).endswith( "-" * (Rota.rota_days_length // 2) ) if worker.name == "worker2": assert Rota.get_worker_shift_list_string(worker).startswith( "-" * (Rota.rota_days_length // 2) ) if worker.name == "worker3": assert Rota.get_worker_shift_list_string(worker).startswith( "-" * (Rota.rota_days_length // 4) ) assert Rota.get_worker_shift_list_string(worker).endswith( "-" * (Rota.rota_days_length // 4) ) assert Rota.results.solver.status == "ok" assert Rota.results.solver.termination_condition == "optimal" Rota.export_rota_to_html("worker_ftes") def test_worker_oops(self): weeks_to_rota = 10 start_date = datetime.date(2022, 3, 7) Rota = RotaBuilder( start_date, weeks_to_rota=weeks_to_rota, ) worker1_oop = [ { "start_date": start_date, "end_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 2), "reason": "First half OOP", } ] worker2_oop = [ { "start_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 2), "end_date": Rota.rota_end_date, "reason": "First half OOP", } ] worker3_start_date = start_date + datetime.timedelta(weeks=weeks_to_rota / 4) worker3_oop = [ { "start_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 2), "end_date": start_date + datetime.timedelta(weeks=3 * weeks_to_rota / 4), "reason": "Quarter oop", } ] worker4_oop = [ { "start_date": start_date, "end_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 4), "reason": "Quarter oop", }, { "start_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 2), "end_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 2) + datetime.timedelta(weeks=weeks_to_rota / 4), "reason": "Quarter oop 2", }, ] workers = [] workers.append( Worker(name="worker1", site="group1", grade=1, fte=100, oop=worker1_oop) ) workers.append( Worker(name="worker2", site="group1", grade=1, fte=100, oop=worker2_oop) ) workers.append( Worker( name="worker3", site="group2", grade=1, fte=100, start_date=worker3_start_date, oop=worker3_oop, ) ) workers.append(Worker(name="worker4", site="group2", grade=1, oop=worker4_oop)) workers.append( Worker(name="worker5", site="group2", grade=1, fte=50, nwd=["Mon", "Tue"]) ) worker6_oop = [ { "start_date": start_date, "end_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 6), "reason": "oop", }, { "start_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 2), "end_date": start_date + datetime.timedelta(weeks=weeks_to_rota / 2) + datetime.timedelta(weeks=weeks_to_rota / 3), "reason": " oop 2", }, ] worker7_oop = [ { "start_date": start_date - datetime.timedelta(weeks=1), "end_date": Rota.rota_end_date, "reason": "oop", }, ] workers.append( Worker(name="worker6", site="group2", grade=1, fte=100, oop=worker6_oop) ) workers.append( Worker(name="worker7", site="group3", grade=1, fte=100, oop=worker7_oop) ) Rota.add_workers(workers) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="d", length=12.5, days=days[:5], balance_offset=10, workers_required=1, ), ) Rota.terminate_on_warning.remove("Worker/no valid shifts") Rota.build_and_solve() # All workers should have an adjusted fte of 50 % for worker in Rota.get_workers(): if worker.site == "group3": assert worker.fte_adj == 0 continue assert worker.fte_adj == pytest.approx(50, 2) if worker.fte == 100: assert worker.proportion_rota_to_work == pytest.approx(0.5, 0.2) assert worker.days_to_work == pytest.approx(35, 1) assert Rota.get_worker_shift_list(worker).count("d") in (8, 9) assert Rota.results.solver.status == "ok" assert Rota.results.solver.termination_condition == "optimal" Rota.export_rota_to_html("worker_oops") def test_worker_pairs(self): Rota = generate_basic_rota() Rota.add_shifts( SingleShift( sites=("group1",), name="a", length=12.5, days=days[:5], balance_offset=20, workers_required=1, ), SingleShift( sites=("group1",), name="b", length=12.5, days=days[:5], balance_offset=20, workers_required=1, ), ) Rota.add_workers( [ Worker(name="worker1", site="group1", grade=1, fte=100, pair="1"), Worker(name="worker2", site="group1", grade=1, fte=100, pair="1"), ] ) Rota.constraint_options["hard_constrain_pair_separation"] = True Rota.build_and_solve() # With two paired workers we shouldn't be able to solve assert Rota.results.solver.status in ("warning", "error") worker3 = Worker(name="worker3", site="group1", grade=1, fte=100) Rota.add_worker(worker3) # An additional worker can take half the shifts Rota.build_and_solve() assert Rota.results.solver.status == "ok" assert Rota.get_workers_total_shifts()["worker3"] == 50 assert ( Rota.get_workers_total_shifts()["worker1"] + Rota.get_workers_total_shifts()["worker2"] == 50 ) worker3.pair = "1" Rota.build_and_solve() assert Rota.results.solver.status in ("warning", "error") worker3.pair = 0 Rota.build_and_solve() Rota.export_rota_to_html("worker_pairs") assert Rota.results.solver.status == "ok" worker4 = Worker(name="worker4", site="group1", grade=1, fte=100, pair="1") Rota.add_worker(worker4) Rota.build_and_solve() Rota.export_rota_to_html("worker_pairs") assert Rota.results.solver.status == "ok" assert Rota.get_workers_total_shifts()["worker3"] == 50 def test_not_available_to_work(self): Rota = generate_basic_rota() Rota.add_shifts( SingleShift( sites=("group1",), name="a", length=12.5, days=days, workers_required=1, ), ) natws = generate_not_available_to_works(Rota.start_date, days=21) natws.append( NotAvailableToWork(date=Rota.rota_end_date - datetime.timedelta(days=1)) ) Rota.add_workers( [ Worker(name="worker1", site="group1", grade=1), Worker( name="worker2", site="group1", grade=1, not_available_to_work=natws ), ] ) Rota.build_and_solve() Rota.export_rota_to_html("test_not_available_to_work") assert Rota.results.solver.status == "ok" assert Rota.get_workers_total_shifts()["worker1"] == 35 assert Rota.get_workers_total_shifts()["worker2"] == 35 assert Rota.get_worker_shift_list_string( Rota.get_worker_by_name("worker2") ).startswith("-" * 21) assert Rota.get_worker_shift_list_string( Rota.get_worker_by_name("worker2") ).endswith("-") class TestWorkerFtes: def test_worker_ftes(self): Rota = generate_basic_rota(workers=2) Rota.add_workers( [ Worker( name="worker3", site="group1", grade=1, fte=60, ), Worker( name="worker4", site="group1", grade=1, fte=40, ), ] ) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:6], workers_required=1, ), ) Rota.build_and_solve() Rota.export_rota_to_html("test_worker_ftes") total_shifts = Rota.get_workers_total_shifts() for worker in Rota.get_workers(): match worker.name: case "worker1": assert worker.get_fte() == 100 assert total_shifts[worker.name] == 20 case "worker2": assert worker.get_fte() == 100 assert total_shifts[worker.name] == 20 case "worker3": assert worker.get_fte() == 60 assert total_shifts[worker.name] == 12 case "worker4": assert worker.get_fte() == 40 assert total_shifts[worker.name] == 8 def test_worker_fte_shift_override(self): Rota = generate_basic_rota(workers=1) Rota.add_workers( [ Worker( name="worker2", site="group1", grade=1, fte=100, shift_fte_overrides={"b": 50} ), ] ) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:2], workers_required=1, ), SingleShift( sites=("group1", "group2"), name="b", length=12.5, days=days[2:5], workers_required=1, ), ) Rota.build_and_solve() Rota.export_rota_to_html("test_worker_fte_shift_override") total_shifts = Rota.get_workers_total_shifts() for worker in Rota.get_workers(): match worker.name: case "worker1": assert worker.get_fte() == 100 assert total_shifts[worker.name] == 30 assert worker.get_fte("b") == 100 case "worker2": assert worker.get_fte() == 100 assert worker.get_fte("b") == 50 assert total_shifts[worker.name] == 20 class TestComplexWorkers: def test_worker_requirement_and_force_block(self): Rota = generate_basic_rota(workers=10, weeks_to_rota=10) wr = [ WorkerRequirement( end_date=Rota.start_date + datetime.timedelta(weeks=5), number=2, ), WorkerRequirement( start_date=Rota.start_date + datetime.timedelta(weeks=5), number=3, ), ] Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:2], workers_required=1, ), SingleShift( sites=("group1", "group2"), name="b", length=12.5, days=days[2:5], workers_required=1, ), SingleShift( sites=("group1", "group2"), name="c", length=12.5, days=days[:4], workers_required=wr, force_as_block=True, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, ] ), SingleShift( sites=("group1", "group2"), name="d", length=12.5, days=days[4:], workers_required=wr, force_as_block=True, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, ] ), ) Rota.build_and_solve() Rota.export_rota_to_html("test_worker_complex") #total_shifts = Rota.get_workers_total_shifts() #for worker in Rota.get_workers(): # match worker.name: # case "worker1": # assert worker.get_fte() == 100 # assert total_shifts[worker.name] == 30 # assert worker.get_fte("b") == 100 # case "worker2": # assert worker.get_fte() == 100 # assert worker.get_fte("b") == 50 # assert total_shifts[worker.name] == 20 #total_shifts = Rota.get_workers_total_shifts() assert Rota.results.solver.status == "ok" for worker in Rota.get_workers(): shift_string = Rota.get_worker_shift_list_string(worker) assert shift_string.count("cccc") == shift_string.count("c") / 4 assert shift_string.count("ddd") == shift_string.count("d") / 3 def test_split_shift(self): Rota = generate_basic_rota(workers=11, weeks_to_rota=22) Rota.set_rota_constraint("max_shifts_per_week", 2) Rota.set_rota_constraint("max_weekend_frequency", 3) Rota.set_rota_constraint("max_shifts_per_month", 4) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:4], workers_required=1, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, {"name": "max_shifts_per_week", "options": 1}, ] ), SingleShift( sites=("group1", "group2"), name="b", length=12.5, days=(days[4], days[6]), workers_required=1, assign_as_block=True, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, ] ), SingleShift( sites=("group1", "group2"), name="c", length=12.5, days=days[5], workers_required=1, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, ] ), ) Rota.build_and_solve() Rota.export_rota_to_html("test_split") assert Rota.results.solver.status == "ok" class TestLocums: def test_locums_basic(self): Rota = generate_basic_rota(workers=2, weeks_to_rota=6) la = [] for date in Rota.get_date_range(end_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): #for date in Rota.get_date_range(): la.append(WorkRequests(date=date, shift="*")) la2 = [] for date in Rota.get_date_range(start_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): #for date in Rota.get_date_range(): la2.append(WorkRequests(date=date, shift="*")) Rota.add_workers( [ Worker(name="worker03", site="group1", locum=False, locum_availability=la, locum_max_shifts=40, locum_max_shifts_per_week=1), Worker(name="worker04", site="group1", locum=False, locum_availability=la2, locum_max_shifts=40, locum_max_shifts_per_week=2), Worker(name="locum1", site="group1", locum=True), Worker( name="locum2", site="group1", locum=True ), ] ) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:4], workers_required=1, constraint=[ #{"name": "pre", "options": 1}, #{"name": "post", "options": 1}, #{"name": "max_shifts_per_week", "options": 1}, ] ), ) Rota.set_rota_constraint("distribute_locum_shifts", True) Rota.set_rota_constraint("maximum_allowed_shift_diff", 0) Rota.set_rota_constraint("maximum_allowed_locum_shifts_per_worker", 4) Rota.build_and_solve() Rota.export_rota_to_html("test_locums_basic", timestamp_filename=False) assert Rota.results.solver.status == "ok" for worker in Rota.get_workers(): assert Rota.get_worker_shift_list(worker).count("a") == 4 def test_locums_nwds(self): Rota = generate_basic_rota(workers=2, weeks_to_rota=6) la = [] for date in Rota.get_date_range(end_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): la.append(WorkRequests(date=date, shift="*")) la2 = [] for date in Rota.get_date_range(start_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): la2.append(WorkRequests(date=date, shift="*")) Rota.add_workers( [ Worker(name="worker03", site="group1", locum=False, locum_availability=la, locum_max_shifts=40, locum_max_shifts_per_week=1, nwds=[NonWorkingDays(day="Tue")]), Worker(name="worker04", site="group1", locum=False, locum_availability=la2, locum_max_shifts=40, locum_max_shifts_per_week=2, nwds=[NonWorkingDays(day="Wed"),NonWorkingDays(day="Thu")]), #Worker(name="worker03", site="group1", locum=False, locum_availability=la, locum_max_shifts=40, locum_max_shifts_per_week=1, nwds=[NonWorkingDays(day="Mon"),NonWorkingDays(day="Tue")]), #Worker(name="worker04", site="group1", locum=False, locum_availability=la2, locum_max_shifts=40, locum_max_shifts_per_week=2, nwds=[NonWorkingDays(day="Wed"),NonWorkingDays(day="Thu")]), Worker(name="locum1", site="group1", locum=True), Worker( name="locum2", site="group1", locum=True ), ] ) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:4], workers_required=1, constraint=[ #{"name": "pre", "options": 1}, #{"name": "post", "options": 1}, #{"name": "max_shifts_per_week", "options": 1}, ] ), ) #Rota.set_rota_constraint("distribute_locum_shifts", True) Rota.set_rota_constraint("maximum_allowed_shift_diff", 0) Rota.set_rota_constraint("maximum_allowed_locum_shifts_per_worker", 4) #Rota.set_rota_constraint("minimum_allowed_locum_shifts_per_worker", 3) Rota.build_and_solve() Rota.export_rota_to_html("test_locums_basic", timestamp_filename=False) assert Rota.results.solver.status == "ok" for worker in Rota.get_workers(): print( Rota.get_worker_shift_list(worker)) assert Rota.get_worker_shift_list(worker).count("a") == 4 if worker.name in ("worker03", "worker04"): assert Rota.get_worker_shift_list(worker, include_locums=True).count("a") > 6 def test_locums_no_availablity(self): Rota = generate_basic_rota(workers=2, weeks_to_rota=5) la = [] #for date in Rota.get_date_range(end_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): # la.append(WorkRequests(date=date, shift="*")) la2 = [] #for date in Rota.get_date_range(start_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): # la2.append(WorkRequests(date=date, shift="*")) Rota.add_workers( [ Worker(name="worker03", site="group1", locum=False, locum_availability=la, locum_max_shifts=40, locum_max_shifts_per_week=1), Worker(name="worker04", site="group1", locum=False, locum_availability=la2, locum_max_shifts=40, locum_max_shifts_per_week=2), Worker(name="locum1", site="group1", locum=True), Worker( name="locum2", site="group1", locum=True ), ] ) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:4], workers_required=1, constraint=[ #{"name": "pre", "options": 1}, #{"name": "post", "options": 1}, #{"name": "max_shifts_per_week", "options": 1}, ] ), ) Rota.set_rota_constraint("distribute_locum_shifts", False) with pytest.raises(WarningTermination): Rota.build_and_solve() def test_locums(self): Rota = generate_basic_rota(workers=7, weeks_to_rota=10) Rota.set_rota_constraint("max_shifts_per_week", 2) Rota.set_rota_constraint("max_weekend_frequency", 3) Rota.set_rota_constraint("max_shifts_per_month", 4) #Rota.set_rota_constraint("balance_shifts", False) Rota.set_rota_constraint("distribute_locum_shifts", True) Rota.set_rota_constraint("maximum_allowed_shift_diff", 1) Rota.set_rota_constraint("maximum_allowed_locum_shifts_per_worker", 6) Rota.terminate_on_warning.remove("Worker/no valid shifts") la = [] for date in Rota.get_date_range(end_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): la.append(WorkRequests(date=date, shift="*")) la2 = [] for date in Rota.get_date_range(start_date=Rota.rota_end_date-datetime.timedelta(weeks=3)): la2.append(WorkRequests(date=date, shift="*")) Rota.add_workers( [ Worker(name="worker08", site="group1", locum=False, locum_availability=la, locum_max_shifts=40, locum_max_shifts_per_week=1), Worker(name="worker09", site="group1", locum=False, locum_availability=la2, locum_max_shifts=40, locum_max_shifts_per_week=2), Worker(name="locum1", site="group1", locum=True), Worker( name="locum2", site="group1", locum=True ), #Worker( # name="extra1", site="locum", locum_availability=la, # locum_max_shifts=40 #), #Worker( # name="extra2", site="locum", locum_availability=la, # locum_max_shifts=4, locum_max_shifts_per_week=1 #), ] ) Rota.add_shifts( SingleShift( sites=("group1", "group2"), name="a", length=12.5, days=days[:4], workers_required=1, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, {"name": "max_shifts_per_week", "options": 1}, ] ), SingleShift( sites=("group1", "group2"), name="b", length=12.5, days=(days[4], days[6]), workers_required=1, assign_as_block=True, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, ] ), SingleShift( sites=("group1", "group2"), name="c", length=12.5, days=days[5], workers_required=1, constraint=[ {"name": "pre", "options": 2}, {"name": "post", "options": 3}, ] ), ) Rota.build_and_solve() Rota.export_rota_to_html("test_locums", timestamp_filename=False) assert Rota.results.solver.status == "ok"