mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 09:03:11 +00:00
470e00affb
* feat(schedule): align the work week with Sunday-Saturday payroll Saturday night stays in the week that ends Saturday, and the first Flex close skips dates already sent. * fix(slack-bot): show the last pay close on Sunday The Monday 7am row for the week that just ended is not written yet, so /oncall pay now falls back to the prior close.
624 lines
27 KiB
Python
624 lines
27 KiB
Python
"""Tests for shared.schedule — the DynamoDB single-table data model."""
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from shared.schedule import (
|
|
DEFAULT_HOLIDAY_MULTIPLIER,
|
|
FALLBACK_EXTENSION,
|
|
ShiftSchedule,
|
|
determine_shift_type,
|
|
week_start,
|
|
)
|
|
|
|
ET = ZoneInfo("America/New_York")
|
|
|
|
# 2026-06-01 is a Monday, so 06-06 is Saturday and 06-07 is Sunday.
|
|
SAT = "2026-06-06"
|
|
SUN = "2026-06-07"
|
|
WED = "2026-06-03"
|
|
HOL = "2026-07-04" # holiday date (a Saturday)
|
|
|
|
|
|
class TestWeekStart:
|
|
def test_sunday_opens_the_week(self):
|
|
sunday = datetime(2026, 6, 7, 15, tzinfo=ET)
|
|
assert week_start(sunday).strftime("%Y-%m-%d") == "2026-06-07"
|
|
|
|
def test_monday_belongs_to_the_sunday_before(self):
|
|
monday = datetime(2026, 6, 8, 8, tzinfo=ET)
|
|
assert week_start(monday).strftime("%Y-%m-%d") == "2026-06-07"
|
|
|
|
def test_saturday_stays_in_the_week_that_started_sunday(self):
|
|
saturday = datetime(2026, 6, 13, 23, tzinfo=ET)
|
|
assert week_start(saturday).strftime("%Y-%m-%d") == "2026-06-07"
|
|
|
|
|
|
class TestDetermineShiftType:
|
|
@pytest.mark.parametrize(
|
|
"dt,expected",
|
|
[
|
|
(datetime(2026, 6, 6, 8, 0, tzinfo=ET), "day"), # Sat 08:00 boundary
|
|
(datetime(2026, 6, 6, 10, 0, tzinfo=ET), "day"), # Sat mid-day
|
|
(datetime(2026, 6, 6, 16, 59, tzinfo=ET), "day"), # Sat just before 17
|
|
(datetime(2026, 6, 6, 17, 0, tzinfo=ET), "night"), # Sat 17:00 boundary
|
|
(datetime(2026, 6, 6, 7, 59, tzinfo=ET), "night"), # Sat before 08
|
|
(datetime(2026, 6, 7, 12, 0, tzinfo=ET), "day"), # Sun mid-day
|
|
(datetime(2026, 6, 3, 12, 0, tzinfo=ET), "night"), # Wed — always night
|
|
(datetime(2026, 6, 3, 9, 0, tzinfo=ET), "night"), # weekday morning
|
|
],
|
|
)
|
|
def test_boundaries(self, dt, expected):
|
|
assert determine_shift_type(dt) == expected
|
|
|
|
def test_defaults_to_now_when_none(self):
|
|
# Should not raise and returns a valid shift type.
|
|
assert determine_shift_type() in ("day", "night")
|
|
|
|
|
|
class TestResolveShift:
|
|
def test_unassigned_is_available(self, schedule):
|
|
assert schedule.resolve_shift(WED, "Wednesday") == (
|
|
FALLBACK_EXTENSION,
|
|
"Available",
|
|
"available",
|
|
)
|
|
|
|
def test_weekly_assignment(self, schedule, seed):
|
|
seed.weekly("Wednesday", "114", "Alice")
|
|
assert schedule.resolve_shift(WED, "Wednesday") == ("114", "Alice", "weekly")
|
|
|
|
def test_weekly_fallback_extension_is_available(self, schedule, seed):
|
|
seed.weekly("Wednesday", FALLBACK_EXTENSION, "Fallback")
|
|
ext, name, source = schedule.resolve_shift(WED, "Wednesday")
|
|
assert (ext, source) == (FALLBACK_EXTENSION, "available")
|
|
|
|
def test_override_takes_precedence_over_weekly(self, schedule, seed):
|
|
seed.weekly("Wednesday", "114", "Alice")
|
|
seed.override(WED, "115", "Bob")
|
|
assert schedule.resolve_shift(WED, "Wednesday") == ("115", "Bob", "override")
|
|
|
|
def test_open_override_is_available(self, schedule, seed):
|
|
seed.weekly("Wednesday", "114", "Alice")
|
|
seed.open_shift(WED)
|
|
ext, name, source = schedule.resolve_shift(WED, "Wednesday")
|
|
assert (ext, name, source) == (FALLBACK_EXTENSION, "Available", "available")
|
|
|
|
def test_weekend_day_shift_uses_day_key(self, schedule, seed):
|
|
seed.weekly("Saturday", "200", "DayPerson", shift_type="day")
|
|
seed.weekly("Saturday", "201", "NightPerson", shift_type="night")
|
|
assert schedule.resolve_shift(SAT, "Saturday", "day") == (
|
|
"200",
|
|
"DayPerson",
|
|
"weekly",
|
|
)
|
|
assert schedule.resolve_shift(SAT, "Saturday", "night") == (
|
|
"201",
|
|
"NightPerson",
|
|
"weekly",
|
|
)
|
|
|
|
def test_day_and_night_overrides_are_independent(self, schedule, seed):
|
|
seed.override(SAT, "200", "DayPerson", shift_type="day")
|
|
# Night shift has no override → available
|
|
assert schedule.resolve_shift(SAT, "Saturday", "day")[0] == "200"
|
|
assert schedule.resolve_shift(SAT, "Saturday", "night")[2] == "available"
|
|
|
|
|
|
class TestOverrides:
|
|
def test_set_and_get_override_night(self, schedule):
|
|
schedule.set_override(WED, "114", "Alice")
|
|
item = schedule.get_override(WED)
|
|
assert item["extension"] == "114" and item["name"] == "Alice"
|
|
|
|
def test_set_and_get_override_day_uses_separate_sk(self, schedule):
|
|
schedule.set_override(SAT, "200", "DayPerson", shift_type="day")
|
|
assert schedule.get_override(SAT, "day")["extension"] == "200"
|
|
# The night SK must be untouched
|
|
assert schedule.get_override(SAT, "night") is None
|
|
|
|
def test_mark_open(self, schedule):
|
|
schedule.set_override(WED, "114", "Alice")
|
|
schedule.mark_open(WED)
|
|
assert schedule.get_override(WED)["extension"] == "OPEN"
|
|
|
|
def test_remove_override_reverts_to_weekly(self, schedule, seed):
|
|
seed.weekly("Wednesday", "114", "Alice")
|
|
schedule.set_override(WED, "115", "Bob")
|
|
schedule.remove_override(WED)
|
|
assert schedule.resolve_shift(WED, "Wednesday") == ("114", "Alice", "weekly")
|
|
|
|
|
|
class TestListOverrides:
|
|
def test_returns_night_and_day_in_range(self, schedule, seed):
|
|
seed.override(WED, "115", "Bob")
|
|
seed.override(SAT, "200", "DayPerson", shift_type="day")
|
|
seed.open_shift(SUN)
|
|
result = schedule.list_overrides(WED, SUN)
|
|
assert result == [
|
|
{"date": WED, "shift_type": "night", "extension": "115", "name": "Bob"},
|
|
{
|
|
"date": SAT,
|
|
"shift_type": "day",
|
|
"extension": "200",
|
|
"name": "DayPerson",
|
|
},
|
|
{"date": SUN, "shift_type": "night", "extension": "OPEN", "name": "Open"},
|
|
]
|
|
|
|
def test_excludes_dates_outside_range(self, schedule, seed):
|
|
seed.override("2026-05-31", "100", "Before")
|
|
seed.override(WED, "115", "Bob")
|
|
seed.override("2026-06-30", "300", "After")
|
|
result = schedule.list_overrides(WED, SUN)
|
|
assert [r["extension"] for r in result] == ["115"]
|
|
|
|
def test_includes_day_override_on_end_date(self, schedule, seed):
|
|
seed.override(SUN, "201", "EndDay", shift_type="day")
|
|
result = schedule.list_overrides(WED, SUN)
|
|
assert result == [
|
|
{"date": SUN, "shift_type": "day", "extension": "201", "name": "EndDay"}
|
|
]
|
|
|
|
def test_empty_when_no_overrides(self, schedule):
|
|
assert schedule.list_overrides(WED, SUN) == []
|
|
|
|
|
|
class TestClaimOpenShift:
|
|
def test_claim_when_no_override_succeeds(self, schedule):
|
|
assert schedule.claim_open_shift(WED, "114", "Alice") is True
|
|
assert schedule.get_override(WED)["extension"] == "114"
|
|
|
|
def test_claim_when_open_succeeds(self, schedule):
|
|
schedule.mark_open(WED)
|
|
assert schedule.claim_open_shift(WED, "114", "Alice") is True
|
|
assert schedule.get_override(WED)["extension"] == "114"
|
|
|
|
def test_claim_when_already_assigned_fails_and_does_not_overwrite(self, schedule):
|
|
schedule.set_override(WED, "115", "Bob")
|
|
assert schedule.claim_open_shift(WED, "114", "Alice") is False
|
|
assert schedule.get_override(WED)["extension"] == "115"
|
|
|
|
|
|
class TestRoster:
|
|
def test_get_roster_and_by_extension(self, schedule, seed):
|
|
seed.roster("114", "Alice")
|
|
seed.roster("115", "Bob")
|
|
assert {e["SK"] for e in schedule.get_roster()} == {"114", "115"}
|
|
assert schedule.get_employee_by_extension("114")["name"] == "Alice"
|
|
assert schedule.get_employee_by_extension("999") is None
|
|
|
|
def test_get_employee_by_slack_id(self, schedule, seed):
|
|
seed.roster("114", "Alice", slack_user_id="U_ALICE")
|
|
seed.roster("115", "Bob")
|
|
assert schedule.get_employee_by_slack_id("U_ALICE")["SK"] == "114"
|
|
assert schedule.get_employee_by_slack_id("U_NOBODY") is None
|
|
|
|
def test_register_user_links_slack_id(self, schedule, seed):
|
|
seed.roster("114", "Alice")
|
|
result = schedule.register_user("U_ALICE", "114")
|
|
assert result["slack_user_id"] == "U_ALICE"
|
|
assert schedule.get_employee_by_extension("114")["slack_user_id"] == "U_ALICE"
|
|
|
|
def test_register_user_unknown_extension_returns_none(self, schedule):
|
|
assert schedule.register_user("U_ALICE", "999") is None
|
|
|
|
def test_add_roster_entry_new_and_duplicate(self, schedule):
|
|
assert schedule.add_roster_entry("114", "Alice") is True
|
|
assert schedule.add_roster_entry("114", "Alice Again") is False
|
|
assert schedule.get_employee_by_extension("114")["name"] == "Alice"
|
|
assert schedule.get_employee_by_extension("114")["slack_user_id"] == ""
|
|
|
|
def test_upsert_roster_entry_creates_row_with_slack_id(self, schedule):
|
|
schedule.upsert_roster_entry("114", "Alice", "U_ALICE")
|
|
emp = schedule.get_employee_by_extension("114")
|
|
assert emp["PK"] == "ROSTER" and emp["SK"] == "114"
|
|
assert emp["name"] == "Alice"
|
|
assert emp["extension"] == "114"
|
|
assert emp["slack_user_id"] == "U_ALICE"
|
|
assert "shift_rate" not in emp
|
|
|
|
def test_upsert_roster_entry_updates_slack_id_and_preserves_shift_rate(
|
|
self, schedule, seed
|
|
):
|
|
seed.roster("114", "Alice", slack_user_id="", shift_rate="75")
|
|
schedule.upsert_roster_entry("114", "Alicia", "U_ALICE")
|
|
emp = schedule.get_employee_by_extension("114")
|
|
assert emp["name"] == "Alicia"
|
|
assert emp["extension"] == "114"
|
|
assert emp["slack_user_id"] == "U_ALICE"
|
|
assert emp["shift_rate"] == "75"
|
|
|
|
def test_upsert_writes_email_and_lookup_is_case_insensitive(self, schedule):
|
|
schedule.upsert_roster_entry(
|
|
"114", "Alice", "U_ALICE", email="Alice@Seahavenind.com"
|
|
)
|
|
emp = schedule.get_employee_by_email("alice@seahavenind.com")
|
|
assert emp["extension"] == "114"
|
|
assert emp["email"] == "Alice@Seahavenind.com"
|
|
|
|
def test_upsert_without_email_preserves_existing_email(self, schedule, seed):
|
|
seed.roster(
|
|
"114", "Alice", slack_user_id="U_OLD", email="alice@seahavenind.com"
|
|
)
|
|
schedule.upsert_roster_entry("114", "Alicia", "U_ALICE")
|
|
emp = schedule.get_employee_by_extension("114")
|
|
assert emp["email"] == "alice@seahavenind.com"
|
|
assert emp["slack_user_id"] == "U_ALICE"
|
|
seed.roster("114", "Alice", slack_user_id="U_ALICE", shift_rate="75")
|
|
assert schedule.add_roster_entry("114", "Overwrite") is False
|
|
emp = schedule.get_employee_by_extension("114")
|
|
assert emp["name"] == "Alice"
|
|
assert emp["slack_user_id"] == "U_ALICE"
|
|
assert emp["shift_rate"] == "75"
|
|
|
|
def test_remove_roster_entry(self, schedule, seed):
|
|
seed.roster("114", "Alice")
|
|
schedule.remove_roster_entry("114")
|
|
assert schedule.get_employee_by_extension("114") is None
|
|
|
|
def test_rename_roster_entry(self, schedule, seed):
|
|
seed.roster("114", "Alice")
|
|
schedule.rename_roster_entry("114", "Alicia")
|
|
assert schedule.get_employee_by_extension("114")["name"] == "Alicia"
|
|
|
|
|
|
class TestRates:
|
|
def test_default_rate_from_config(self, schedule, seed):
|
|
seed.config(shift_rate="50")
|
|
assert schedule.get_shift_rate() == 50.0
|
|
|
|
def test_default_rate_missing_is_zero(self, schedule):
|
|
assert schedule.get_shift_rate() == 0.0
|
|
|
|
def test_per_employee_rate_overrides_default(self, schedule, seed):
|
|
seed.config(shift_rate="50")
|
|
seed.roster("114", "Alice", shift_rate="75")
|
|
assert schedule.get_shift_rate("114") == 75.0
|
|
# Employee without a custom rate falls back to default
|
|
seed.roster("115", "Bob")
|
|
assert schedule.get_shift_rate("115") == 50.0
|
|
|
|
def test_set_default_shift_rate(self, schedule, seed):
|
|
seed.config(shift_rate="50")
|
|
schedule.set_default_shift_rate(60)
|
|
assert schedule.get_shift_rate() == 60.0
|
|
|
|
def test_set_employee_shift_rate(self, schedule, seed):
|
|
seed.roster("114", "Alice")
|
|
schedule.set_employee_shift_rate("114", 90)
|
|
assert schedule.get_shift_rate("114") == 90.0
|
|
|
|
|
|
class TestPayAndConfigAndPosts:
|
|
def test_pay_record_round_trip(self, schedule):
|
|
schedule.save_pay_record("2026-05-25", {"breakdown": [], "totals": {}})
|
|
rec = schedule.get_pay_record("2026-05-25")
|
|
assert rec["PK"] == "PAY" and rec["SK"] == "2026-05-25"
|
|
|
|
def test_get_pay_record_missing(self, schedule):
|
|
assert schedule.get_pay_record("1999-01-01") is None
|
|
|
|
def test_admin_users(self, schedule, seed):
|
|
seed.config(admin_users=["U_ADMIN"])
|
|
assert schedule.get_admin_users() == ["U_ADMIN"]
|
|
|
|
def test_admin_users_default_empty(self, schedule):
|
|
assert schedule.get_admin_users() == []
|
|
|
|
def test_schedule_post_round_trip(self, schedule):
|
|
schedule.save_schedule_post("C123", "111.222", "2026-06-01")
|
|
post = schedule.get_schedule_post("C123")
|
|
assert post["message_ts"] == "111.222"
|
|
assert schedule.get_schedule_post("C_NONE") is None
|
|
|
|
|
|
def test_shiftschedule_uses_env_table_name(monkeypatch, dynamodb_table):
|
|
# Constructed with no arg → reads SHIFT_TABLE.
|
|
assert ShiftSchedule().table.name == "afterhours-shifts"
|
|
|
|
|
|
class TestSwaps:
|
|
REQ = {"extension": "114", "name": "Alice", "slack_user_id": "U_ALICE"}
|
|
TGT = {"extension": "115", "name": "Bob", "slack_user_id": "U_BOB"}
|
|
|
|
def test_create_and_get_pending_swap(self, schedule):
|
|
schedule.create_pending_swap(WED, "night", self.REQ, self.TGT, 1_900_000_000)
|
|
swap = schedule.get_swap(WED)
|
|
assert swap["status"] == "pending"
|
|
assert swap["requester_ext"] == "114" and swap["requester_slack"] == "U_ALICE"
|
|
assert swap["target_ext"] == "115" and swap["target_slack"] == "U_BOB"
|
|
assert swap["expires_at"] == 1_900_000_000
|
|
assert swap["created_at"] # stamped
|
|
|
|
def test_get_missing_swap(self, schedule):
|
|
assert schedule.get_swap(WED) is None
|
|
|
|
def test_create_supersedes_existing(self, schedule):
|
|
schedule.create_pending_swap(WED, "night", self.REQ, self.TGT, 1)
|
|
other = {"extension": "116", "name": "Carol", "slack_user_id": "U_CAROL"}
|
|
schedule.create_pending_swap(WED, "night", self.REQ, other, 2)
|
|
swap = schedule.get_swap(WED)
|
|
assert swap["target_ext"] == "116" # overwritten, only one record
|
|
|
|
def test_day_and_night_swaps_independent(self, schedule):
|
|
schedule.create_pending_swap(SAT, "day", self.REQ, self.TGT, 1)
|
|
assert schedule.get_swap(SAT, "day")["target_ext"] == "115"
|
|
assert schedule.get_swap(SAT, "night") is None
|
|
|
|
def test_mark_verified(self, schedule):
|
|
schedule.create_pending_swap(WED, "night", self.REQ, self.TGT, 1)
|
|
schedule.mark_swap_verified(WED)
|
|
swap = schedule.get_swap(WED)
|
|
assert swap["status"] == "verified" and swap["accepted_at"]
|
|
|
|
def test_clear_swap(self, schedule):
|
|
schedule.create_pending_swap(WED, "night", self.REQ, self.TGT, 1)
|
|
schedule.clear_swap(WED)
|
|
assert schedule.get_swap(WED) is None
|
|
|
|
|
|
class TestHolidayConfig:
|
|
def test_holiday_multiplier_default(self, schedule):
|
|
assert schedule.get_holiday_multiplier() == DEFAULT_HOLIDAY_MULTIPLIER
|
|
|
|
def test_holiday_multiplier_override(self, schedule, seed):
|
|
seed.config(holiday_multiplier=Decimal("2"))
|
|
assert schedule.get_holiday_multiplier() == Decimal("2")
|
|
|
|
def test_holiday_queue_default_and_override(self, schedule, seed):
|
|
assert schedule.get_holiday_queue() == "802"
|
|
seed.config(holiday_queue="900")
|
|
assert schedule.get_holiday_queue() == "900"
|
|
|
|
def test_ivr_number_default_and_override(self, schedule, seed):
|
|
assert schedule.get_ivr_number() == "800"
|
|
seed.config(ivr_number="700")
|
|
assert schedule.get_ivr_number() == "700"
|
|
|
|
def test_captured_ivr_routes_roundtrip_and_clear(self, schedule, seed):
|
|
seed.config()
|
|
assert schedule.get_captured_ivr_routes() is None
|
|
routes = {
|
|
"key0": {"To": "Extension", "Number": "101", "External": ""},
|
|
"timeout": {"To": "Extension", "Number": "102", "External": ""},
|
|
}
|
|
schedule.set_captured_ivr_routes(routes)
|
|
assert schedule.get_captured_ivr_routes() == routes
|
|
schedule.set_captured_ivr_routes(None)
|
|
assert schedule.get_captured_ivr_routes() is None
|
|
|
|
|
|
class TestHolidays:
|
|
def test_create_and_get(self, schedule):
|
|
assert schedule.create_holiday(HOL, 2, "Independence Day", "U_ADMIN") is True
|
|
h = schedule.get_holiday(HOL)
|
|
assert h["slots"] == 2
|
|
assert h["label"] == "Independence Day"
|
|
assert h["created_by"] == "U_ADMIN"
|
|
assert h["assignees"] == {}
|
|
assert h["activated"] is False
|
|
assert h["multiplier"] == DEFAULT_HOLIDAY_MULTIPLIER
|
|
|
|
def test_create_uses_config_multiplier(self, schedule, seed):
|
|
seed.config(holiday_multiplier=Decimal("3"))
|
|
schedule.create_holiday(HOL, 1, "Holiday")
|
|
assert schedule.get_holiday(HOL)["multiplier"] == Decimal("3")
|
|
|
|
def test_create_explicit_multiplier_overrides(self, schedule):
|
|
schedule.create_holiday(HOL, 1, "Holiday", multiplier="2.5")
|
|
assert schedule.get_holiday(HOL)["multiplier"] == Decimal("2.5")
|
|
|
|
def test_create_duplicate_fails(self, schedule):
|
|
assert schedule.create_holiday(HOL, 2, "First") is True
|
|
assert schedule.create_holiday(HOL, 5, "Second") is False
|
|
# Original record untouched.
|
|
assert schedule.get_holiday(HOL)["label"] == "First"
|
|
|
|
def test_get_missing(self, schedule):
|
|
assert schedule.get_holiday(HOL) is None
|
|
|
|
def test_claim_slot_succeeds(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
assert schedule.claim_holiday_slot(HOL, "114", "Alice") is True
|
|
h = schedule.get_holiday(HOL)
|
|
assert h["assignees"]["114"]["name"] == "Alice"
|
|
assert h["assignees"]["114"]["claimed_at"]
|
|
|
|
def test_claim_fills_multiple_slots(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
assert schedule.claim_holiday_slot(HOL, "114", "Alice") is True
|
|
assert schedule.claim_holiday_slot(HOL, "115", "Bob") is True
|
|
assert set(schedule.get_holiday(HOL)["assignees"]) == {"114", "115"}
|
|
|
|
def test_claim_full_slot_rejected(self, schedule):
|
|
schedule.create_holiday(HOL, 1, "Holiday")
|
|
assert schedule.claim_holiday_slot(HOL, "114", "Alice") is True
|
|
assert schedule.claim_holiday_slot(HOL, "115", "Bob") is False
|
|
assert set(schedule.get_holiday(HOL)["assignees"]) == {"114"}
|
|
|
|
def test_claim_duplicate_rejected(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
assert schedule.claim_holiday_slot(HOL, "114", "Alice") is True
|
|
assert schedule.claim_holiday_slot(HOL, "114", "Alice Dup") is False
|
|
# The original claim is preserved unchanged.
|
|
assert schedule.get_holiday(HOL)["assignees"]["114"]["name"] == "Alice"
|
|
|
|
def test_claim_missing_holiday_fails(self, schedule):
|
|
assert schedule.claim_holiday_slot(HOL, "114", "Alice") is False
|
|
|
|
def test_release_slot_succeeds(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
schedule.claim_holiday_slot(HOL, "114", "Alice")
|
|
assert schedule.release_holiday_slot(HOL, "114") is True
|
|
assert schedule.get_holiday(HOL)["assignees"] == {}
|
|
|
|
def test_release_not_assigned_fails(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
assert schedule.release_holiday_slot(HOL, "114") is False
|
|
|
|
def test_release_frees_slot_for_reclaim(self, schedule):
|
|
schedule.create_holiday(HOL, 1, "Holiday")
|
|
schedule.claim_holiday_slot(HOL, "114", "Alice")
|
|
assert schedule.claim_holiday_slot(HOL, "115", "Bob") is False # full
|
|
schedule.release_holiday_slot(HOL, "114")
|
|
assert schedule.claim_holiday_slot(HOL, "115", "Bob") is True
|
|
|
|
def test_swap_assignee_succeeds(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
schedule.claim_holiday_slot(HOL, "114", "Alice")
|
|
assert schedule.swap_holiday_assignee(HOL, "114", "115", "Bob") is True
|
|
assignees = schedule.get_holiday(HOL)["assignees"]
|
|
assert "114" not in assignees
|
|
assert assignees["115"]["name"] == "Bob"
|
|
|
|
def test_swap_from_not_assigned_fails(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
assert schedule.swap_holiday_assignee(HOL, "114", "115", "Bob") is False
|
|
|
|
def test_swap_to_already_assigned_fails(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
schedule.claim_holiday_slot(HOL, "114", "Alice")
|
|
schedule.claim_holiday_slot(HOL, "115", "Bob")
|
|
assert schedule.swap_holiday_assignee(HOL, "114", "115", "Bob") is False
|
|
# Both still present.
|
|
assert set(schedule.get_holiday(HOL)["assignees"]) == {"114", "115"}
|
|
|
|
def test_set_activated(self, schedule):
|
|
schedule.create_holiday(HOL, 1, "Holiday")
|
|
schedule.set_holiday_activated(HOL, True)
|
|
assert schedule.get_holiday(HOL)["activated"] is True
|
|
|
|
def test_remove_holiday(self, schedule):
|
|
schedule.create_holiday(HOL, 1, "Holiday")
|
|
schedule.remove_holiday(HOL)
|
|
assert schedule.get_holiday(HOL) is None
|
|
|
|
|
|
class TestShiftContext:
|
|
def test_available_when_nothing_assigned(self, schedule):
|
|
ctx = schedule.get_shift_context(WED, "Wednesday")
|
|
assert ctx["kind"] == "available"
|
|
assert ctx["assignees"] == []
|
|
assert ctx["slots"] == 1 and ctx["open_slots"] == 1
|
|
assert ctx["multiplier"] == Decimal("1")
|
|
assert ctx["label"] == ""
|
|
|
|
def test_weekly_context(self, schedule, seed):
|
|
seed.weekly("Wednesday", "114", "Alice")
|
|
ctx = schedule.get_shift_context(WED, "Wednesday")
|
|
assert ctx["kind"] == "weekly"
|
|
assert ctx["assignees"] == [{"extension": "114", "name": "Alice"}]
|
|
assert ctx["open_slots"] == 0
|
|
|
|
def test_override_context(self, schedule, seed):
|
|
seed.weekly("Wednesday", "114", "Alice")
|
|
seed.override(WED, "115", "Bob")
|
|
ctx = schedule.get_shift_context(WED, "Wednesday")
|
|
assert ctx["kind"] == "override"
|
|
assert ctx["assignees"] == [{"extension": "115", "name": "Bob"}]
|
|
|
|
def test_open_override_is_available(self, schedule, seed):
|
|
seed.weekly("Wednesday", "114", "Alice")
|
|
seed.open_shift(WED)
|
|
ctx = schedule.get_shift_context(WED, "Wednesday")
|
|
assert ctx["kind"] == "available" and ctx["assignees"] == []
|
|
|
|
def test_holiday_priority_over_override_and_weekly(self, schedule, seed):
|
|
# Day-shift holiday beats both override and weekly.
|
|
seed.weekly("Saturday", "200", "DayPerson", shift_type="day")
|
|
seed.override(HOL, "201", "OverridePerson", shift_type="day")
|
|
schedule.create_holiday(HOL, 2, "Independence Day")
|
|
schedule.claim_holiday_slot(HOL, "114", "Alice")
|
|
schedule.claim_holiday_slot(HOL, "115", "Bob")
|
|
ctx = schedule.get_shift_context(HOL, "Saturday", "day")
|
|
assert ctx["kind"] == "holiday"
|
|
assert ctx["label"] == "Independence Day"
|
|
assert ctx["slots"] == 2 and ctx["open_slots"] == 0
|
|
assert ctx["multiplier"] == DEFAULT_HOLIDAY_MULTIPLIER
|
|
assert {a["extension"] for a in ctx["assignees"]} == {"114", "115"}
|
|
|
|
def test_holiday_open_slots_reported(self, schedule):
|
|
schedule.create_holiday(HOL, 3, "Holiday")
|
|
schedule.claim_holiday_slot(HOL, "114", "Alice")
|
|
ctx = schedule.get_shift_context(HOL, "Saturday", "day")
|
|
assert ctx["slots"] == 3 and ctx["open_slots"] == 2
|
|
|
|
def test_holiday_only_applies_to_day_shift(self, schedule):
|
|
# Night shift on a holiday date ignores the holiday record.
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
ctx = schedule.get_shift_context(HOL, "Saturday", "night")
|
|
assert ctx["kind"] == "available"
|
|
|
|
def test_resolve_shift_wrapper_holiday(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
schedule.claim_holiday_slot(HOL, "114", "Alice")
|
|
assert schedule.resolve_shift(HOL, "Saturday", "day") == (
|
|
"114",
|
|
"Alice",
|
|
"holiday",
|
|
)
|
|
|
|
def test_resolve_shift_wrapper_empty_holiday_available(self, schedule):
|
|
schedule.create_holiday(HOL, 2, "Holiday")
|
|
assert schedule.resolve_shift(HOL, "Saturday", "day") == (
|
|
FALLBACK_EXTENSION,
|
|
"Available",
|
|
"available",
|
|
)
|
|
|
|
|
|
class TestPickupRequests:
|
|
REQ = {"extension": "114", "name": "Alice", "slack_user_id": "U_ALICE"}
|
|
|
|
def test_create_and_get(self, schedule):
|
|
schedule.create_pickup_request(WED, "night", self.REQ, 1_900_000_000)
|
|
req = schedule.get_pickup_request(WED, "night", "114")
|
|
assert req["status"] == "pending"
|
|
assert req["requester_ext"] == "114"
|
|
assert req["requester_name"] == "Alice"
|
|
assert req["requester_slack"] == "U_ALICE"
|
|
assert req["shift_type"] == "night"
|
|
assert req["is_holiday"] is False
|
|
assert req["expires_at"] == 1_900_000_000
|
|
assert req["created_at"]
|
|
|
|
def test_create_holiday_pickup(self, schedule):
|
|
schedule.create_pickup_request(
|
|
HOL, "day", self.REQ, 1_900_000_000, is_holiday=True
|
|
)
|
|
req = schedule.get_pickup_request(HOL, "day", "114")
|
|
assert req["is_holiday"] is True and req["shift_type"] == "day"
|
|
|
|
def test_get_missing(self, schedule):
|
|
assert schedule.get_pickup_request(WED, "night", "114") is None
|
|
|
|
def test_requests_keyed_per_requester(self, schedule):
|
|
other = {"extension": "115", "name": "Bob", "slack_user_id": "U_BOB"}
|
|
schedule.create_pickup_request(WED, "night", self.REQ, 1)
|
|
schedule.create_pickup_request(WED, "night", other, 2)
|
|
assert schedule.get_pickup_request(WED, "night", "114")["expires_at"] == 1
|
|
assert schedule.get_pickup_request(WED, "night", "115")["expires_at"] == 2
|
|
|
|
def test_day_and_night_independent(self, schedule):
|
|
schedule.create_pickup_request(SAT, "day", self.REQ, 1)
|
|
assert schedule.get_pickup_request(SAT, "day", "114") is not None
|
|
assert schedule.get_pickup_request(SAT, "night", "114") is None
|
|
|
|
def test_clear(self, schedule):
|
|
schedule.create_pickup_request(WED, "night", self.REQ, 1)
|
|
schedule.clear_pickup_request(WED, "night", "114")
|
|
assert schedule.get_pickup_request(WED, "night", "114") is None
|
|
|
|
def test_approve_is_atomic_first_wins(self, schedule):
|
|
schedule.create_pickup_request(WED, "night", self.REQ, 1)
|
|
# First approval wins the pending->approved transition; a second
|
|
# concurrent approval of the same request loses (no double-claim).
|
|
assert schedule.approve_pickup_request(WED, "night", "114") is True
|
|
assert schedule.approve_pickup_request(WED, "night", "114") is False
|
|
|
|
def test_approve_missing_request(self, schedule):
|
|
assert schedule.approve_pickup_request(WED, "night", "114") is False
|