mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 04:33:10 +00:00
3a26343cb7
* Add pytest suite and wire it into CI Stands up the first automated tests for the repo (151 tests) and turns on the CI test step. - Lift slack-bot handlers out of create_app() closures to module level so they're unit-testable; create_app is now a thin Bolt-wiring layer. No behavior change (handler entrypoints and create_app signature unchanged). - tests/ mirrors src/: shared layer (schedule, blocks, 3CX client, ring_scheduler, secrets) + all four Lambdas (pay math, drop/swap/pick/ admin/register/rate, pickup button, roster sync, queue scheduler). - All boundaries mocked: DynamoDB/SES/Secrets via moto, 3CX HTTP via responses, Slack via fakes, time via freezegun. No real network/AWS. - pyproject.toml pytest config (pythonpath=src/shared, importlib mode); per-package conftest loads each app.py under a unique name to avoid the four-app.py collision. tests/requirements.txt for test-only deps. - ci.yaml: run-tests: true (reusable workflow auto-installs deps) and lint the tests dir too. - README Testing section. Closes #85 * Add least-privilege permissions block to CI workflow Resolves the CodeQL actions/missing-workflow-permissions alert: the CI workflow now restricts GITHUB_TOKEN to contents: read (it only checks out, lints, and runs tests). * Stop logging extension numbers in 3CX queue updates Resolves 3 high CodeQL py/clear-text-logging-sensitive-data alerts: the queue/ring-group forwarding logs no longer include the routed extension values (closed/holiday/extension). Non-sensitive context (resource id, queue number) is retained.
220 lines
9 KiB
Python
220 lines
9 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"
|