mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 10:13:11 +00:00
eedbdfd8d1
/oncall swap no longer reassigns immediately. It now writes a pending SWAP record and DMs the target Accept/Decline buttons; the shift only moves once they accept. - schedule.py: create_pending_swap / get_swap / mark_swap_verified / clear_swap (PK=SWAP, date/shift SK mirroring OVERRIDE, status + timestamps + expires_at for TTL). A new request supersedes a prior pending one. - app.py: _handle_swap creates the pending swap + DMs the target (requires the target be Slack-linked; rejects self-swap). New module-level handle_swap_accept / handle_swap_decline + two @app.action registrations. Accept writes the override, repoints 3CX when it's the active shift, marks the swap verified, notifies the channel + requester. Decline clears it and DMs the requester. Lazy expiry: accept is rejected once the shift has started (_shift_start/_shift_started). - blocks.py: build_swap_request_blocks (Accept/Decline) + build_swap_resolved_blocks. - template.yaml: enable DynamoDB TTL on expires_at so abandoned pending swaps self-clean. - tests: swap schedule methods, swap blocks, rewritten test_handle_swap (pending + DM, no immediate override), new test_swap_accept_decline. 174 passed. - README: swap behavior + SWAP item type + TTL. The verified SWAP status is what #84 (24h drop guard) will query. Closes #83
260 lines
11 KiB
Python
260 lines
11 KiB
Python
"""Tests for shared.schedule — the DynamoDB single-table data model."""
|
|
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
|
|
from shared.schedule import FALLBACK_EXTENSION, ShiftSchedule, determine_shift_type
|
|
|
|
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"
|
|
|
|
|
|
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 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"
|
|
|
|
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
|