mirror of
https://github.com/Sea-Haven-Industries/afterhours-shift-manager.git
synced 2026-09-30 14:53:12 +00:00
Adam Moussa
4a4ed2a653
Holiday day-shifts (08:00-17:00 ET) with N slots and 1.5x pay. A new afterhours-holiday-router Lambda, fired by per-holiday EventBridge Scheduler one-offs, repoints IVR 800 (key-0 + no-input/timeout) to holiday queue 802 and sets 802's membership to the day's assignees (ext 100 fallback when unfilled), reverting at 17:00. Pickups after a shift starts go through an admin Approve/Deny flow for both regular and holiday shifts. Pay (weekly post + /oncall pay) shows holiday rates distinctly. Adds HOLIDAY and PICKUP_REQUEST DynamoDB record types, scheduler IAM scoped to holiday-* schedules with conditioned PassRole, and the holiday-router function with a 60-day log group and error alarm.
135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
"""Shared pytest fixtures: AWS env, a moto-backed DynamoDB table, a ShiftSchedule
|
|
bound to it, and a seeding helper for the single-table data model."""
|
|
|
|
import boto3
|
|
import pytest
|
|
from moto import mock_aws
|
|
|
|
TABLE_NAME = "afterhours-shifts"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def aws_env(monkeypatch):
|
|
"""Fake AWS credentials + the env vars the handlers read.
|
|
|
|
Autouse so no test can accidentally reach real AWS. 3CX env vars are left
|
|
unset by default, so the slack-bot's _update_3cx_routing short-circuits
|
|
unless a test sets them explicitly.
|
|
"""
|
|
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
|
|
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
|
|
monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
|
|
monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")
|
|
monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
|
|
monkeypatch.setenv("SHIFT_TABLE", TABLE_NAME)
|
|
monkeypatch.setenv("SHIFT_CHANNEL", "C_TEST")
|
|
monkeypatch.delenv("QUEUE_NUMBER", raising=False)
|
|
monkeypatch.delenv("TCX_SECRET_PREFIX", raising=False)
|
|
|
|
|
|
def _create_table(dynamodb):
|
|
table = dynamodb.create_table(
|
|
TableName=TABLE_NAME,
|
|
KeySchema=[
|
|
{"AttributeName": "PK", "KeyType": "HASH"},
|
|
{"AttributeName": "SK", "KeyType": "RANGE"},
|
|
],
|
|
AttributeDefinitions=[
|
|
{"AttributeName": "PK", "AttributeType": "S"},
|
|
{"AttributeName": "SK", "AttributeType": "S"},
|
|
],
|
|
BillingMode="PAY_PER_REQUEST",
|
|
)
|
|
table.wait_until_exists()
|
|
return table
|
|
|
|
|
|
@pytest.fixture
|
|
def dynamodb_table(aws_env):
|
|
"""A mocked `afterhours-shifts` table (PK/SK, PAY_PER_REQUEST). The moto
|
|
context stays open for the duration of the test."""
|
|
with mock_aws():
|
|
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
|
|
yield _create_table(dynamodb)
|
|
|
|
|
|
@pytest.fixture
|
|
def schedule(dynamodb_table):
|
|
"""A ShiftSchedule bound to the mocked table."""
|
|
from shared.schedule import ShiftSchedule
|
|
|
|
return ShiftSchedule()
|
|
|
|
|
|
class Seeder:
|
|
"""Thin helper to put single-table items matching schedule.py's key model."""
|
|
|
|
def __init__(self, table):
|
|
self.table = table
|
|
|
|
def roster(self, ext, name, slack_user_id="", **extra):
|
|
self.table.put_item(
|
|
Item={
|
|
"PK": "ROSTER",
|
|
"SK": ext,
|
|
"name": name,
|
|
"extension": ext,
|
|
"slack_user_id": slack_user_id,
|
|
**extra,
|
|
}
|
|
)
|
|
|
|
def weekly(self, day_name, ext, name, shift_type="night"):
|
|
sk = f"{day_name}-Day" if shift_type == "day" else day_name
|
|
self.table.put_item(
|
|
Item={"PK": "WEEKLY", "SK": sk, "extension": ext, "name": name}
|
|
)
|
|
|
|
def override(self, date_str, ext, name, shift_type="night"):
|
|
sk = f"{date_str}-DAY" if shift_type == "day" else date_str
|
|
self.table.put_item(
|
|
Item={"PK": "OVERRIDE", "SK": sk, "extension": ext, "name": name}
|
|
)
|
|
|
|
def open_shift(self, date_str, shift_type="night"):
|
|
sk = f"{date_str}-DAY" if shift_type == "day" else date_str
|
|
self.table.put_item(
|
|
Item={"PK": "OVERRIDE", "SK": sk, "extension": "OPEN", "name": "Open"}
|
|
)
|
|
|
|
def config(self, **kwargs):
|
|
self.table.put_item(Item={"PK": "CONFIG", "SK": "CONFIG", **kwargs})
|
|
|
|
def holiday(self, date_str, slots=2, label="Holiday", assignees=None, **extra):
|
|
from decimal import Decimal
|
|
|
|
self.table.put_item(
|
|
Item={
|
|
"PK": "HOLIDAY",
|
|
"SK": date_str,
|
|
"slots": slots,
|
|
"assignees": assignees or {},
|
|
"multiplier": Decimal("1.5"),
|
|
"label": label,
|
|
"created_at": "2026-06-01T00:00:00-04:00",
|
|
"created_by": "",
|
|
"activated": False,
|
|
"schedule_names": [],
|
|
**extra,
|
|
}
|
|
)
|
|
|
|
def schedule_post(self, channel, message_ts, week_start="2026-06-01"):
|
|
self.table.put_item(
|
|
Item={
|
|
"PK": "SCHEDULE_POST",
|
|
"SK": channel,
|
|
"message_ts": message_ts,
|
|
"week_start": week_start,
|
|
}
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def seed(dynamodb_table):
|
|
return Seeder(dynamodb_table)
|