Compare commits

..

1 commit

Author SHA1 Message Date
dependabot[bot]
aa5a0ecc85
Update boto3 requirement from >=1.43.11 to >=1.43.15 in /src/shared
Updates the requirements on [boto3](https://github.com/boto/boto3) to permit the latest version.
- [Release notes](https://github.com/boto/boto3/releases)
- [Commits](https://github.com/boto/boto3/compare/1.43.11...1.43.15)

---
updated-dependencies:
- dependency-name: boto3
  dependency-version: 1.43.15
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
2026-05-27 02:42:23 +00:00
36 changed files with 635 additions and 3005 deletions

View file

@ -3,12 +3,8 @@ on:
pull_request:
branches: [main]
permissions:
contents: read
jobs:
ci:
uses: Sea-Haven-Industries/.github/.github/workflows/ci-python-sam.yaml@main
with:
source-dirs: "src/slack-bot src/weekly-post src/roster-sync src/ring-scheduler src/shared/shared tests"
run-tests: true
source-dirs: "src/slack-bot src/weekly-post src/roster-sync src/ring-scheduler src/shared/shared"

2
.gitignore vendored
View file

@ -2,8 +2,6 @@ __pycache__/
*.pyc
.aws-sam/
venv/
.venv/
.pytest_cache/
.env
samconfig.toml
output.json

View file

@ -15,8 +15,8 @@ The weekly schedule post is updated live when shifts change, and the previous we
| `/oncall` | Show this week's schedule |
| `/oncall next` | Show next week's schedule |
| `/oncall pick <date>` | Pick up an available shift |
| `/oncall drop <date>` | Drop your shift (marks it available) — blocked within 24h of shift start; swap or ask an admin instead |
| `/oncall swap <date> @person` | Request a swap — the other person gets an Accept/Decline DM and the shift only moves once they accept |
| `/oncall drop <date>` | Drop your shift (marks it available) |
| `/oncall swap <date> @person` | Hand your shift to someone else |
| `/oncall register <ext>` | Link your Slack account to your phone extension |
| `/oncall roster` | Show all employees and their link status |
| `/oncall pay` | Show last week's bonus pay summary |
@ -67,7 +67,6 @@ src/
roster-sync/ Daily 3CX roster sync
ring-scheduler/ 3CX queue routing updates
shared/ Lambda Layer (schedule, blocks, 3CX client, secrets)
tests/ pytest suite (mirrors src/, one dir per Lambda + shared)
```
### DynamoDB Schema
@ -79,15 +78,10 @@ Single table with `PK` / `SK` keys:
| `ROSTER` | `<extension>` | Employee: name, extension, slack_user_id |
| `WEEKLY` | `<DayName>` | Default weekly schedule: extension, name |
| `OVERRIDE` | `<YYYY-MM-DD>` | Date override from pickup/drop (or `OPEN`) |
| `SWAP` | `<YYYY-MM-DD>` | Pending/verified swap request: requester, target, status, `expires_at` (TTL) |
| `SCHEDULE_POST` | `<channel_id>` | Current schedule message timestamp |
| `PAY` | `<YYYY-MM-DD>` | Weekly pay record (Monday date key) |
| `CONFIG` | `CONFIG` | Settings: shift_rate, fallback_extension, admin_users |
Weekend day-shift rows use a `-DAY` suffix on the SK (e.g. `OVERRIDE` / `2026-04-05-DAY`). The table has TTL enabled on `expires_at` so abandoned pending swaps self-clean.
**Swap flow:** `/oncall swap` writes a `pending` `SWAP` record and DMs the target Accept/Decline buttons; it does **not** reassign the shift. On Accept, the override is written, 3CX is repointed if it's the active shift, and the record is marked `verified`. On Decline (or once the shift has started) the request is dropped and the shift stays with the original owner.
### Secrets Manager
| Secret | Description |
@ -109,23 +103,4 @@ sam build
sam deploy
```
## Testing
Unit tests use `pytest` with all external boundaries mocked — DynamoDB / SES /
Secrets Manager via `moto`, 3CX HTTP via `responses`, Slack via fakes, and time
via `freezegun`. No test touches the network or real AWS.
```bash
python -m venv .venv && source .venv/bin/activate
pip install -r tests/requirements.txt # test-only deps
pip install -r src/slack-bot/requirements.txt \
-r src/weekly-post/requirements.txt \
-r src/shared/requirements.txt # runtime deps the imports need
pytest
```
Each Lambda has its own `app.py`, so the per-package `conftest.py` loads each one
under a unique module name (importlib mode) to avoid collisions. CI runs the same
suite on every PR via the org `ci-python-sam` workflow (`run-tests: true`).
See [SETUP.md](SETUP.md) for full deployment and Slack app creation instructions.

View file

@ -1,7 +0,0 @@
[tool.pytest.ini_options]
# `src/shared` on the path makes the `shared` layer package importable as it is at
# runtime. Each Lambda's own `app.py` is loaded under a unique name by the
# per-package conftest (importlib mode) to avoid the four-`app.py` collision.
pythonpath = ["src/shared"]
testpaths = ["tests"]
addopts = "--import-mode=importlib"

View file

@ -1,2 +1,2 @@
boto3>=1.43.19
boto3>=1.43.15
requests>=2.34.2

View file

@ -149,54 +149,6 @@ def build_shift_change_message(
return [{"type": "section", "text": {"type": "mrkdwn", "text": text}}]
def build_swap_request_blocks(
requester_slack: str, date_str: str, shift_type: str = "night"
) -> list[dict]:
"""Build the interactive Accept / Decline message DMed to a swap target."""
dt = datetime.strptime(date_str, "%Y-%m-%d")
day_label = dt.strftime("%A, %b %-d")
type_label = (
f" ({SHIFT_LABELS.get(shift_type, shift_type)})"
if dt.strftime("%A") in WEEKEND_DAYS
else ""
)
action_suffix = "_day" if shift_type == "day" else ""
return [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": (
f"<@{requester_slack}> wants you to cover the "
f"*{day_label}*{type_label} shift. Accept to take it on."
),
},
},
{
"type": "actions",
"elements": [
{
"type": "button",
"text": {"type": "plain_text", "text": "Accept"},
"action_id": f"swap_accept_{date_str}{action_suffix}",
"style": "primary",
},
{
"type": "button",
"text": {"type": "plain_text", "text": "Decline"},
"action_id": f"swap_decline_{date_str}{action_suffix}",
"style": "danger",
},
],
},
]
def build_swap_resolved_blocks(text: str) -> list[dict]:
"""A plain section used to replace a swap DM's buttons once it's resolved."""
return [{"type": "section", "text": {"type": "mrkdwn", "text": text}}]
def build_help_blocks(is_admin: bool = False) -> list[dict]:
"""Build help message blocks."""
text = (
@ -204,7 +156,7 @@ def build_help_blocks(is_admin: bool = False) -> list[dict]:
"`/oncall` — Show the two-week schedule\n"
"`/oncall next` — Show the following two weeks\n"
"`/oncall pick <date>` — Pick up a shift\n"
"`/oncall drop <date>` — Drop your shift (marks it open; locked within 24h of start — swap instead)\n"
"`/oncall drop <date>` — Drop your shift (marks it open)\n"
"`/oncall swap <date> @person` — Swap your shift with someone\n"
"`/oncall register <extension>` — Link your Slack account to your extension\n"
"`/oncall pay` — Show last week's pay summary\n"

View file

@ -27,5 +27,5 @@ def update_queue_routing(
closed=extension,
holiday=extension,
)
logger.info("Updated queue %s forwarding", queue_number)
logger.info("Updated queue %s to forward to Ext %s", queue_number, extension)
return {"extension": extension, "queue": queue_number}

View file

@ -130,60 +130,6 @@ class ShiftSchedule:
sk = f"{date_str}-DAY" if shift_type == "day" else date_str
self.table.delete_item(Key={"PK": "OVERRIDE", "SK": sk})
# ── Swap requests ───────────────────────────────────────────────────
def create_pending_swap(
self,
date_str: str,
shift_type: str,
requester: dict,
target: dict,
expires_at: int,
) -> None:
"""Create (or supersede) a pending swap request for a shift.
One swap per shift (unique SK), so a new request overwrites any prior
pending one. ``expires_at`` is an epoch timestamp used for DynamoDB TTL.
"""
sk = f"{date_str}-DAY" if shift_type == "day" else date_str
self.table.put_item(
Item={
"PK": "SWAP",
"SK": sk,
"shift_type": shift_type,
"status": "pending",
"requester_ext": requester["extension"],
"requester_name": requester["name"],
"requester_slack": requester.get("slack_user_id", ""),
"target_ext": target["extension"],
"target_name": target["name"],
"target_slack": target.get("slack_user_id", ""),
"created_at": datetime.now(EASTERN).isoformat(),
"expires_at": expires_at,
}
)
def get_swap(self, date_str: str, shift_type: str = "night") -> dict | None:
sk = f"{date_str}-DAY" if shift_type == "day" else date_str
resp = self.table.get_item(Key={"PK": "SWAP", "SK": sk})
return resp.get("Item")
def mark_swap_verified(self, date_str: str, shift_type: str = "night") -> None:
sk = f"{date_str}-DAY" if shift_type == "day" else date_str
self.table.update_item(
Key={"PK": "SWAP", "SK": sk},
UpdateExpression="SET #s = :v, accepted_at = :a",
ExpressionAttributeNames={"#s": "status"},
ExpressionAttributeValues={
":v": "verified",
":a": datetime.now(EASTERN).isoformat(),
},
)
def clear_swap(self, date_str: str, shift_type: str = "night") -> None:
sk = f"{date_str}-DAY" if shift_type == "day" else date_str
self.table.delete_item(Key={"PK": "SWAP", "SK": sk})
# ── Resolution ──────────────────────────────────────────────────────
def resolve_shift(

View file

@ -102,7 +102,13 @@ class ThreeCXClient:
json=payload,
)
resp.raise_for_status()
logger.info("Updated %s %s forwarding", resource, resource_id)
logger.info(
"Updated %s %s: closed->Ext %s, holiday->Ext %s",
resource,
resource_id,
closed,
holiday,
)
return resp.status_code
def get_ring_group(self, extension_number: str) -> dict:

File diff suppressed because it is too large Load diff

View file

@ -1,2 +1,2 @@
slack_sdk>=3.42.0,<4.0
boto3>=1.43.19
boto3>=1.43.11

View file

@ -53,9 +53,6 @@ Resources:
KeyType: HASH
- AttributeName: SK
KeyType: RANGE
TimeToLiveSpecification:
AttributeName: expires_at
Enabled: true
# --- Slack Bot Lambda ---
SlackBotFunction:

View file

@ -1,116 +0,0 @@
"""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 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)

View file

@ -1,7 +0,0 @@
# Test-only dependencies. The CI reusable workflow (ci-python-sam.yaml) installs
# every requirements.txt it finds when run-tests is true, so this file is picked
# up automatically alongside each Lambda's runtime requirements.
pytest>=8.0
moto[dynamodb,ses,secretsmanager]>=5.0
responses>=0.25
freezegun>=1.5

View file

@ -1,22 +0,0 @@
"""Load src/ring-scheduler/app.py under a unique module name."""
import importlib.util
import pathlib
import sys
import pytest
_ROOT = pathlib.Path(__file__).resolve().parents[2]
def _load(name, relpath):
spec = importlib.util.spec_from_file_location(name, _ROOT / relpath)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
@pytest.fixture
def ringscheduler_app():
return _load("ringscheduler_app", "src/ring-scheduler/app.py")

View file

@ -1,61 +0,0 @@
"""Tests for the ring-scheduler Lambda handler."""
from unittest.mock import MagicMock
import pytest
from freezegun import freeze_time
from shared.schedule import FALLBACK_EXTENSION
# Monday 2026-06-01 08:00 ET — weekday valid_hours is {8}, so the guard passes
# without force; active shift is night.
MON_0800 = "2026-06-01 12:00:00"
@pytest.fixture
def env(monkeypatch):
monkeypatch.setenv("QUEUE_NUMBER", "800")
monkeypatch.setenv("TCX_SECRET_PREFIX", "afterhours-shift-manager/3cx-")
@pytest.fixture
def routing(ringscheduler_app, monkeypatch):
spy = MagicMock(name="update_queue_routing", return_value={"queue": "800"})
monkeypatch.setattr(ringscheduler_app, "update_queue_routing", spy)
monkeypatch.setattr(ringscheduler_app, "get_secret", lambda _id: "secret")
return spy
@freeze_time(MON_0800)
def test_override_extension_bypasses_schedule(ringscheduler_app, env, routing):
result = ringscheduler_app.handler({"override_extension": "199"}, None)
assert routing.call_args.kwargs["extension"] == "199"
assert result["shift_type"] == "night"
@freeze_time(MON_0800)
def test_resolves_assigned_extension(ringscheduler_app, schedule, seed, env, routing):
seed.weekly("Monday", "114", "Alice")
ringscheduler_app.handler({}, None)
assert routing.call_args.kwargs["extension"] == "114"
assert routing.call_args.kwargs["queue_number"] == "800"
@freeze_time(MON_0800)
def test_unassigned_uses_fallback(ringscheduler_app, schedule, seed, env, routing):
ringscheduler_app.handler({}, None)
assert routing.call_args.kwargs["extension"] == FALLBACK_EXTENSION
@freeze_time(MON_0800)
def test_routing_failure_returns_error(ringscheduler_app, schedule, env, routing):
routing.side_effect = RuntimeError("3CX down")
result = ringscheduler_app.handler({"override_extension": "199"}, None)
assert result["error"] is True
@freeze_time("2026-06-01 13:00:00") # 09:00 ET — not a valid scheduler hour
def test_skips_on_wrong_hour(ringscheduler_app, env, routing):
result = ringscheduler_app.handler({}, None)
assert result == {"skipped": True}
routing.assert_not_called()

View file

@ -1,22 +0,0 @@
"""Load src/roster-sync/app.py under a unique module name."""
import importlib.util
import pathlib
import sys
import pytest
_ROOT = pathlib.Path(__file__).resolve().parents[2]
def _load(name, relpath):
spec = importlib.util.spec_from_file_location(name, _ROOT / relpath)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
@pytest.fixture
def rostersync_app():
return _load("rostersync_app", "src/roster-sync/app.py")

View file

@ -1,91 +0,0 @@
"""Tests for the roster-sync Lambda handler."""
from unittest.mock import MagicMock
import pytest
@pytest.fixture
def env(monkeypatch):
monkeypatch.setenv("TCX_SECRET_PREFIX", "afterhours-shift-manager/3cx-")
@pytest.fixture
def fake_3cx(rostersync_app, monkeypatch):
"""Patch ThreeCXClient + get_secret. Returns a setter for group members."""
client = MagicMock(name="ThreeCXClient")
monkeypatch.setattr(rostersync_app, "ThreeCXClient", MagicMock(return_value=client))
monkeypatch.setattr(rostersync_app, "get_secret", lambda _id: "secret")
def set_members(members):
client.get_group_members.return_value = members
return set_members
def test_adds_removes_and_preserves_links(
rostersync_app, schedule, seed, env, fake_3cx
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.roster("115", "Bob", slack_user_id="U_BOB") # departs
fake_3cx(
[
{"Number": "114", "MemberName": "Alice", "Type": "Extension"},
{"Number": "116", "MemberName": "New Person", "Type": "Extension"},
{
"Number": "117",
"MemberName": "Voicemail",
"Type": "Extension",
}, # excluded
{
"Number": "118",
"MemberName": "Reception",
"Type": "RingGroup",
}, # excluded
]
)
result = rostersync_app.handler({"force": True}, None)
# 116 added, 115 removed, 117/118 ignored.
assert any("116" in a for a in result["added"])
assert any("115" in r for r in result["removed"])
assert schedule.get_employee_by_extension("116")["name"] == "New Person"
assert schedule.get_employee_by_extension("115") is None
assert schedule.get_employee_by_extension("117") is None
# Existing link preserved.
assert schedule.get_employee_by_extension("114")["slack_user_id"] == "U_ALICE"
def test_renames_changed_member_preserving_link(
rostersync_app, schedule, seed, env, fake_3cx
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
fake_3cx([{"Number": "114", "MemberName": "Alicia", "Type": "Extension"}])
result = rostersync_app.handler({"force": True}, None)
assert any("114" in u for u in result["updated"])
emp = schedule.get_employee_by_extension("114")
assert emp["name"] == "Alicia"
assert emp["slack_user_id"] == "U_ALICE"
def test_empty_3cx_result_does_not_wipe_roster(
rostersync_app, schedule, seed, env, fake_3cx
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
fake_3cx([])
result = rostersync_app.handler({"force": True}, None)
assert result["removed"] == []
assert schedule.get_employee_by_extension("114") is not None
def test_skips_when_wrong_hour_and_not_forced(rostersync_app, env, fake_3cx):
import freezegun
# 12:00 UTC = 08:00 ET, not 6am → skip.
with freezegun.freeze_time("2026-06-01 12:00:00"):
assert rostersync_app.handler({}, None) == {"skipped": True}

View file

@ -1,159 +0,0 @@
"""Tests for shared.blocks — Block Kit builders."""
from freezegun import freeze_time
from shared.blocks import (
build_help_blocks,
build_pay_summary_blocks,
build_roster_blocks,
build_shift_change_message,
build_swap_request_blocks,
build_swap_resolved_blocks,
build_week_schedule,
)
def _all_action_ids(blocks):
ids = []
for b in blocks:
if b.get("type") == "actions":
ids.extend(e["action_id"] for e in b["elements"])
return ids
class TestBuildWeekSchedule:
@freeze_time("2026-06-01 12:00:00") # Monday
def test_header_and_section_present(self, schedule):
blocks = build_week_schedule(schedule)
assert blocks[0]["type"] == "header"
assert "After-Hours Schedule" in blocks[0]["text"]["text"]
assert blocks[1]["type"] == "section"
@freeze_time("2026-06-01 12:00:00")
def test_all_available_produces_pickup_buttons(self, schedule):
# Empty schedule → every shift is available → pickup buttons exist,
# including a weekend day button with the _day suffix.
blocks = build_week_schedule(schedule)
action_ids = _all_action_ids(blocks)
assert "pickup_2026-06-03" in action_ids # Wednesday night
assert "pickup_2026-06-06_day" in action_ids # Saturday day shift
assert "pickup_2026-06-06" in action_ids # Saturday night shift
@freeze_time("2026-06-01 12:00:00")
def test_assigned_shift_has_no_pickup_button(self, schedule, seed):
seed.weekly("Wednesday", "114", "Alice")
blocks = build_week_schedule(schedule)
assert "pickup_2026-06-03" not in _all_action_ids(blocks)
assert "Alice (Ext 114)" in blocks[1]["text"]["text"]
class TestBuildShiftChangeMessage:
def test_picked_up_weekday(self):
blocks = build_shift_change_message(
"U1", "2026-06-03", "picked_up", "114", "Alice"
)
text = blocks[0]["text"]["text"]
assert "<@U1>" in text and "picked up" in text and "Ext 114" in text
# Weekday → no (Day/Night) label
assert "(Night" not in text
def test_dropped_shows_available(self):
blocks = build_shift_change_message(
"U1", "2026-06-03", "dropped", "114", "Alice"
)
assert "Available" in blocks[0]["text"]["text"]
def test_swapped_text(self):
blocks = build_shift_change_message("U2", "2026-06-03", "swapped", "115", "Bob")
assert "swapped" in blocks[0]["text"]["text"]
def test_weekend_includes_shift_label(self):
blocks = build_shift_change_message(
"U1", "2026-06-06", "picked_up", "200", "Alice", shift_type="day"
)
assert "Day (8am" in blocks[0]["text"]["text"]
class TestBuildPaySummaryBlocks:
def test_renders_breakdown_and_totals(self):
breakdown = [
{
"day": "Mon",
"date_label": "Jun 1",
"name": "Alice",
"extension": "114",
"rate": 50.0,
}
]
totals = {
"Alice": {"shifts": 1, "total": 50.0, "extension": "114", "rate": 50.0}
}
blocks = build_pay_summary_blocks("Jun 1 to Jun 7", breakdown, totals)
assert blocks[0]["type"] == "header"
assert "Jun 1 to Jun 7" in blocks[0]["text"]["text"]
text = blocks[1]["text"]["text"]
assert "Alice" in text and "$50.00" in text and "1 shift" in text
class TestBuildSwapRequestBlocks:
def _action_ids(self, blocks):
return [
e["action_id"]
for b in blocks
if b["type"] == "actions"
for e in b["elements"]
]
def test_weekday_request_has_accept_decline(self):
blocks = build_swap_request_blocks("U_REQ", "2026-06-03", "night")
assert "<@U_REQ>" in blocks[0]["text"]["text"]
ids = self._action_ids(blocks)
assert ids == ["swap_accept_2026-06-03", "swap_decline_2026-06-03"]
def test_weekend_day_request_uses_day_suffix_and_label(self):
blocks = build_swap_request_blocks("U_REQ", "2026-06-06", "day")
assert "Day (8am" in blocks[0]["text"]["text"]
ids = self._action_ids(blocks)
assert ids == ["swap_accept_2026-06-06_day", "swap_decline_2026-06-06_day"]
def test_button_styles(self):
elements = build_swap_request_blocks("U_REQ", "2026-06-03", "night")[1][
"elements"
]
assert elements[0]["style"] == "primary" # Accept
assert elements[1]["style"] == "danger" # Decline
class TestBuildSwapResolvedBlocks:
def test_renders_text_no_buttons(self):
blocks = build_swap_resolved_blocks("All done.")
assert blocks == [
{"type": "section", "text": {"type": "mrkdwn", "text": "All done."}}
]
class TestBuildHelpBlocks:
def test_non_admin_excludes_admin_section(self):
text = build_help_blocks(is_admin=False)[0]["text"]["text"]
assert "Admin Commands" not in text
def test_admin_includes_admin_section(self):
text = build_help_blocks(is_admin=True)[0]["text"]["text"]
assert "Admin Commands" in text
class TestBuildRosterBlocks:
def test_empty_roster(self):
text = build_roster_blocks([])[0]["text"]["text"]
assert "No employees" in text
def test_linked_and_unlinked(self):
roster = [
{"SK": "115", "name": "Bob", "slack_user_id": ""},
{"SK": "114", "name": "Alice", "slack_user_id": "U_ALICE"},
]
text = build_roster_blocks(roster)[0]["text"]["text"]
# Sorted by extension → Alice (114) appears before Bob (115)
assert text.index("Alice") < text.index("Bob")
assert "<@U_ALICE>" in text
assert "_not linked_" in text

View file

@ -1,43 +0,0 @@
"""Tests for shared.ring_scheduler.update_queue_routing (ThreeCXClient faked)."""
import shared.ring_scheduler as ring_scheduler
class FakeThreeCXClient:
instances = []
def __init__(self, domain, auth_mode, client_id, client_secret):
self.domain = domain
self.auth_mode = auth_mode
self.client_id = client_id
self.client_secret = client_secret
self.forwarding = None
FakeThreeCXClient.instances.append(self)
def get_queue(self, number):
self.queue_number = number
return {"Id": 42, "Number": number}
def update_queue_forwarding(self, queue_id, closed, holiday):
self.forwarding = {"queue_id": queue_id, "closed": closed, "holiday": holiday}
return 200
def test_update_queue_routing_points_queue_at_extension(monkeypatch):
FakeThreeCXClient.instances = []
monkeypatch.setattr(ring_scheduler, "ThreeCXClient", FakeThreeCXClient)
result = ring_scheduler.update_queue_routing(
extension="114",
queue_number="800",
domain="test.3cx.us",
client_id="cid",
client_secret="secret",
)
assert result == {"extension": "114", "queue": "800"}
client = FakeThreeCXClient.instances[0]
assert client.auth_mode == "oauth"
assert client.queue_number == "800"
# Both closed and holiday routes point at the on-call extension.
assert client.forwarding == {"queue_id": 42, "closed": "114", "holiday": "114"}

View file

@ -1,260 +0,0 @@
"""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

View file

@ -1,32 +0,0 @@
"""Tests for shared.secrets — Secrets Manager fetch + client caching."""
import boto3
import pytest
from moto import mock_aws
import shared.secrets as secrets
@pytest.fixture(autouse=True)
def _reset_client():
# The module caches a module-level client; reset around each test so a
# client created under one moto context doesn't leak into another.
secrets._client = None
yield
secrets._client = None
def test_get_secret_returns_secret_string():
with mock_aws():
sm = boto3.client("secretsmanager", region_name="us-east-1")
sm.create_secret(
Name="afterhours-shift-manager/3cx-domain", SecretString="x.3cx.us"
)
assert secrets.get_secret("afterhours-shift-manager/3cx-domain") == "x.3cx.us"
def test_get_client_is_cached():
with mock_aws():
first = secrets._get_client()
second = secrets._get_client()
assert first is second

View file

@ -1,113 +0,0 @@
"""Tests for shared.three_cx_client — auth flows and XAPI calls (HTTP mocked)."""
import responses
from shared.three_cx_client import ThreeCXClient
BASE = "https://test.3cx.us"
def _stub_oauth():
responses.add(
responses.POST,
f"{BASE}/connect/token",
json={"access_token": "tok-oauth"},
status=200,
)
@responses.activate
def test_oauth_authentication_sets_bearer_header():
_stub_oauth()
client = ThreeCXClient(
domain="test.3cx.us",
auth_mode="oauth",
client_id="cid",
client_secret="secret",
)
assert client.session.headers["Authorization"] == "Bearer tok-oauth"
@responses.activate
def test_user_authentication_extracts_nested_token():
responses.add(
responses.POST,
f"{BASE}/webclient/api/Login/GetAccessToken",
json={"Token": {"access_token": "tok-user"}},
status=200,
)
client = ThreeCXClient(domain="test.3cx.us", username="u", password="p")
assert client.session.headers["Authorization"] == "Bearer tok-user"
@responses.activate
def test_user_authentication_missing_token_raises():
import pytest
responses.add(
responses.POST,
f"{BASE}/webclient/api/Login/GetAccessToken",
json={"nope": True},
status=200,
)
with pytest.raises(ValueError):
ThreeCXClient(domain="test.3cx.us", username="u", password="p")
@responses.activate
def test_get_group_members_resolves_group_then_members():
_stub_oauth()
responses.add(
responses.GET,
f"{BASE}/xapi/v1/Groups",
json={"value": [{"Id": 5, "Name": "DEFAULT"}]},
status=200,
)
responses.add(
responses.GET,
f"{BASE}/xapi/v1/Groups(5)/Members",
json={"value": [{"Number": "114", "MemberName": "Alice", "Type": "Extension"}]},
status=200,
)
client = ThreeCXClient(
domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
)
members = client.get_group_members("DEFAULT")
assert members == [{"Number": "114", "MemberName": "Alice", "Type": "Extension"}]
@responses.activate
def test_get_group_members_unknown_group_returns_empty():
_stub_oauth()
responses.add(
responses.GET, f"{BASE}/xapi/v1/Groups", json={"value": []}, status=200
)
client = ThreeCXClient(
domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
)
assert client.get_group_members("NOPE") == []
@responses.activate
def test_get_queue_and_update_queue_forwarding():
_stub_oauth()
responses.add(
responses.GET,
f"{BASE}/xapi/v1/Queues/Pbx.GetByNumber(number='800')",
json={"Id": 7, "Number": "800"},
status=200,
)
patched = responses.add(responses.PATCH, f"{BASE}/xapi/v1/Queues(7)", status=200)
client = ThreeCXClient(
domain="test.3cx.us", auth_mode="oauth", client_id="c", client_secret="s"
)
queue = client.get_queue("800")
assert queue["Id"] == 7
status = client.update_queue_forwarding(queue_id=7, closed="114", holiday="114")
assert status == 200
# The forwarding payload routes both closed and holiday to the extension.
import json
body = json.loads(patched.calls[0].request.body)
assert body["OutOfOfficeRoute"]["Route"]["Number"] == "114"
assert body["HolidaysRoute"]["Route"]["Number"] == "114"

View file

@ -1,53 +0,0 @@
"""Load src/slack-bot/app.py under a unique module name so it doesn't collide
with the other Lambdas' app.py modules in sys.modules, plus common test doubles."""
import importlib.util
import pathlib
import sys
from unittest.mock import MagicMock
import pytest
_ROOT = pathlib.Path(__file__).resolve().parents[2]
def _load(name, relpath):
spec = importlib.util.spec_from_file_location(name, _ROOT / relpath)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
@pytest.fixture
def slackbot_app():
return _load("slackbot_app", "src/slack-bot/app.py")
@pytest.fixture
def respond():
return MagicMock(name="respond")
@pytest.fixture
def client():
return MagicMock(name="client")
@pytest.fixture
def routing_spy(slackbot_app, monkeypatch):
"""Replace the 3CX routing call with a spy so tests can assert on it
without touching the network."""
spy = MagicMock(name="_update_3cx_routing")
monkeypatch.setattr(slackbot_app, "_update_3cx_routing", spy)
return spy
@pytest.fixture
def text_of():
"""Helper: pull the `text=` kwarg from a mock's most recent call."""
def _get(mock):
return mock.call_args.kwargs.get("text", "")
return _get

View file

@ -1,90 +0,0 @@
"""Tests for slack-bot _handle_admin."""
from freezegun import freeze_time
from shared.schedule import FALLBACK_EXTENSION
MON = "2026-06-01 12:00:00"
def _admin(slackbot_app, schedule, text, respond, client):
slackbot_app._handle_admin(respond, schedule, "U_ADMIN", text, True, client, None)
def test_non_admin_rejected(slackbot_app, schedule, respond, client, text_of):
slackbot_app._handle_admin(
respond, schedule, "U_USER", "admin open today", False, client, None
)
assert "restricted" in text_of(respond).lower()
def test_admin_help(slackbot_app, schedule, respond, client, text_of):
_admin(slackbot_app, schedule, "admin", respond, client)
assert "Admin Commands" in text_of(respond)
@freeze_time(MON)
def test_override(slackbot_app, schedule, seed, respond, client, routing_spy, text_of):
seed.roster("114", "Alice")
_admin(slackbot_app, schedule, "admin override today 114", respond, client)
assert schedule.get_override("2026-06-01")["extension"] == "114"
assert "Override set" in text_of(respond)
routing_spy.assert_called_once_with("114")
def test_override_unknown_extension(slackbot_app, schedule, respond, client, text_of):
_admin(slackbot_app, schedule, "admin override today 999", respond, client)
assert "not found" in text_of(respond).lower()
@freeze_time(MON)
def test_open(slackbot_app, schedule, seed, respond, client, routing_spy, text_of):
seed.weekly("Monday", "114", "Alice")
_admin(slackbot_app, schedule, "admin open today", respond, client)
assert schedule.get_override("2026-06-01")["extension"] == "OPEN"
assert "marked as open" in text_of(respond).lower()
routing_spy.assert_called_once_with(FALLBACK_EXTENSION)
@freeze_time(MON)
def test_clear(slackbot_app, schedule, seed, respond, client, routing_spy, text_of):
seed.override("2026-06-01", "115", "Bob")
_admin(slackbot_app, schedule, "admin clear today", respond, client)
assert schedule.get_override("2026-06-01") is None
assert "cleared" in text_of(respond).lower()
def test_roster_add(slackbot_app, schedule, respond, client, text_of):
_admin(slackbot_app, schedule, "admin roster add 116 New Person", respond, client)
assert schedule.get_employee_by_extension("116")["name"] == "New Person"
assert "Added" in text_of(respond)
def test_roster_add_duplicate(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("116", "Existing")
_admin(slackbot_app, schedule, "admin roster add 116 New Person", respond, client)
assert "already exists" in text_of(respond).lower()
def test_roster_remove(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("114", "Alice")
_admin(slackbot_app, schedule, "admin roster remove 114", respond, client)
assert schedule.get_employee_by_extension("114") is None
assert "Removed" in text_of(respond)
def test_roster_remove_unknown(slackbot_app, schedule, respond, client, text_of):
_admin(slackbot_app, schedule, "admin roster remove 999", respond, client)
assert "not found" in text_of(respond).lower()
def test_roster_rename(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("114", "Alice")
_admin(slackbot_app, schedule, "admin roster rename 114 Alicia", respond, client)
assert schedule.get_employee_by_extension("114")["name"] == "Alicia"
assert "Renamed" in text_of(respond)
def test_unknown_subcommand(slackbot_app, schedule, respond, client, text_of):
_admin(slackbot_app, schedule, "admin frobnicate", respond, client)
assert "Unknown admin command" in text_of(respond)

View file

@ -1,142 +0,0 @@
"""Tests for slack-bot _handle_drop, including the 24h drop lock (#84)."""
from freezegun import freeze_time
# Monday 2026-06-01 08:00 ET.
MON_0800 = "2026-06-01 12:00:00"
# Monday 2026-06-01 18:00 ET — past 17:00 (so tomorrow's night shift is <24h away).
MON_1800 = "2026-06-01 22:00:00"
# Friday 2026-06-05.
FRI_0600 = "2026-06-05 10:00:00" # 06:00 ET
FRI_0900 = "2026-06-05 13:00:00" # 09:00 ET
@freeze_time(MON_0800)
def test_drop_outside_24h_marks_open(
slackbot_app, schedule, seed, respond, client, routing_spy, text_of
):
# Tomorrow's night shift starts 06-02 17:00; now is 06-01 08:00 → ~33h away.
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Tuesday", "114", "Alice")
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop tomorrow", "C1", client, None
)
assert schedule.get_override("2026-06-02")["extension"] == "OPEN"
assert "dropped" in text_of(respond)
client.chat_postMessage.assert_called_once()
routing_spy.assert_not_called()
@freeze_time(MON_0800)
def test_drop_today_blocked(
slackbot_app, schedule, seed, respond, client, routing_spy, text_of
):
# A same-day shift is always inside the 24h lock.
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Monday", "114", "Alice")
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop today", "C1", client, None
)
assert "under 24 hours" in text_of(respond)
assert schedule.get_override("2026-06-01") is None # not opened
routing_spy.assert_not_called()
client.chat_postMessage.assert_not_called()
@freeze_time(MON_1800)
def test_drop_blocked_once_within_24h_of_start(
slackbot_app, schedule, seed, respond, client, text_of
):
# At 06-01 18:00, tomorrow's 06-02 17:00 shift is <24h away.
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Tuesday", "114", "Alice")
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop tomorrow", "C1", client, None
)
assert "under 24 hours" in text_of(respond)
assert schedule.get_override("2026-06-02") is None
@freeze_time(FRI_0600)
def test_drop_weekend_day_shift_outside_24h(
slackbot_app, schedule, seed, respond, client, text_of
):
# Saturday day shift starts 06-06 08:00; now Fri 06:00 → >24h.
seed.roster("200", "Alice", slack_user_id="U_ALICE")
seed.weekly("Saturday", "200", "Alice", shift_type="day")
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop saturday", "C1", client, None
)
assert "dropped" in text_of(respond)
assert schedule.get_override("2026-06-06", "day")["extension"] == "OPEN"
@freeze_time(FRI_0900)
def test_drop_weekend_day_shift_within_24h(
slackbot_app, schedule, seed, respond, client, text_of
):
# Now Fri 09:00 → Saturday 08:00 day shift is <24h away.
seed.roster("200", "Alice", slack_user_id="U_ALICE")
seed.weekly("Saturday", "200", "Alice", shift_type="day")
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop saturday", "C1", client, None
)
assert "under 24 hours" in text_of(respond)
assert schedule.get_override("2026-06-06", "day") is None
@freeze_time(MON_0800)
def test_admin_open_bypasses_24h_lock(
slackbot_app, schedule, seed, respond, client, routing_spy, text_of
):
# Admins can still open a same-day shift — the lock only applies to user drop.
seed.weekly("Monday", "114", "Alice")
slackbot_app._handle_admin(
respond, schedule, "U_ADMIN", "admin open today", True, client, None
)
assert schedule.get_override("2026-06-01")["extension"] == "OPEN"
assert "marked as open" in text_of(respond).lower()
@freeze_time(MON_0800)
def test_drop_not_your_shift(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Monday", "115", "Bob") # Bob is on shift, not Alice
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop today", "C1", client, None
)
assert "not your shift" in text_of(respond).lower()
@freeze_time(MON_0800)
def test_drop_past_date(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Monday", "114", "Alice")
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop 2026-05-01", "C1", client, None
)
assert "past" in text_of(respond).lower()
@freeze_time(MON_0800)
def test_drop_unregistered(slackbot_app, schedule, respond, client, text_of):
slackbot_app._handle_drop(
respond, schedule, "U_NOBODY", "drop today", "C1", client, None
)
assert "not registered" in text_of(respond).lower()
@freeze_time(MON_0800)
def test_drop_bad_date(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Monday", "114", "Alice")
slackbot_app._handle_drop(
respond, schedule, "U_ALICE", "drop notaday", "C1", client, None
)
assert "couldn't parse" in text_of(respond).lower()
@freeze_time(MON_0800)
def test_drop_usage(slackbot_app, schedule, respond, client, text_of):
slackbot_app._handle_drop(respond, schedule, "U_ALICE", "drop", "C1", client, None)
assert "Usage" in text_of(respond)

View file

@ -1,55 +0,0 @@
"""Tests for slack-bot _handle_pick."""
from freezegun import freeze_time
MON = "2026-06-01 12:00:00"
@freeze_time(MON)
def test_pick_open_shift_today(
slackbot_app, schedule, seed, respond, client, routing_spy, text_of
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
# Monday night is unassigned (available) → pickable.
slackbot_app._handle_pick(
respond, schedule, "U_ALICE", "pick today", "C1", client, None
)
assert schedule.get_override("2026-06-01")["extension"] == "114"
assert "picked up" in text_of(respond).lower()
routing_spy.assert_called_once_with("114")
@freeze_time(MON)
def test_pick_already_covered(
slackbot_app, schedule, seed, respond, client, routing_spy, text_of
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Monday", "115", "Bob") # already covered
slackbot_app._handle_pick(
respond, schedule, "U_ALICE", "pick today", "C1", client, None
)
assert "already covered" in text_of(respond).lower()
routing_spy.assert_not_called()
@freeze_time(MON)
def test_pick_past_date(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
slackbot_app._handle_pick(
respond, schedule, "U_ALICE", "pick 2026-05-01", "C1", client, None
)
assert "past" in text_of(respond).lower()
@freeze_time(MON)
def test_pick_unregistered(slackbot_app, schedule, respond, client, text_of):
slackbot_app._handle_pick(
respond, schedule, "U_NOBODY", "pick today", "C1", client, None
)
assert "not registered" in text_of(respond).lower()
@freeze_time(MON)
def test_pick_usage(slackbot_app, schedule, respond, client, text_of):
slackbot_app._handle_pick(respond, schedule, "U_ALICE", "pick", "C1", client, None)
assert "Usage" in text_of(respond)

View file

@ -1,64 +0,0 @@
"""Tests for slack-bot _handle_register and _handle_rate."""
class TestRegister:
def test_register_links_account(
self, slackbot_app, schedule, seed, respond, text_of
):
seed.roster("114", "Alice")
slackbot_app._handle_register(respond, schedule, "U_ALICE", "register 114")
assert "Linked" in text_of(respond)
assert schedule.get_employee_by_extension("114")["slack_user_id"] == "U_ALICE"
def test_register_unknown_extension(self, slackbot_app, schedule, respond, text_of):
slackbot_app._handle_register(respond, schedule, "U_ALICE", "register 999")
assert "not found" in text_of(respond).lower()
def test_register_usage(self, slackbot_app, schedule, respond, text_of):
slackbot_app._handle_register(respond, schedule, "U_ALICE", "register")
assert "Usage" in text_of(respond)
class TestRate:
def test_show_rates(self, slackbot_app, schedule, seed, respond, text_of):
seed.config(shift_rate="50")
seed.roster("114", "Alice", shift_rate="75")
slackbot_app._handle_rate(respond, schedule, "rate")
text = text_of(respond)
assert "$50.00" in text and "Alice" in text and "$75.00" in text
def test_show_rates_no_custom(self, slackbot_app, schedule, seed, respond, text_of):
seed.config(shift_rate="50")
slackbot_app._handle_rate(respond, schedule, "rate")
assert "No per-person rates" in text_of(respond)
def test_set_default(self, slackbot_app, schedule, seed, respond, text_of):
seed.config(shift_rate="50")
slackbot_app._handle_rate(respond, schedule, "rate default 60")
assert schedule.get_shift_rate() == 60.0
assert "$60.00" in text_of(respond)
def test_set_default_invalid_amount(self, slackbot_app, schedule, respond, text_of):
slackbot_app._handle_rate(respond, schedule, "rate default abc")
assert "Invalid amount" in text_of(respond)
def test_set_default_usage(self, slackbot_app, schedule, respond, text_of):
slackbot_app._handle_rate(respond, schedule, "rate default")
assert "Usage" in text_of(respond)
def test_set_per_employee(self, slackbot_app, schedule, seed, respond, text_of):
seed.roster("114", "Alice")
slackbot_app._handle_rate(respond, schedule, "rate 114 90")
assert schedule.get_shift_rate("114") == 90.0
assert "Alice" in text_of(respond) and "$90.00" in text_of(respond)
def test_set_per_employee_unknown(self, slackbot_app, schedule, respond, text_of):
slackbot_app._handle_rate(respond, schedule, "rate 999 90")
assert "not found" in text_of(respond).lower()
def test_set_per_employee_strips_dollar_sign(
self, slackbot_app, schedule, seed, respond
):
seed.roster("114", "Alice")
slackbot_app._handle_rate(respond, schedule, "rate 114 $90")
assert schedule.get_shift_rate("114") == 90.0

View file

@ -1,139 +0,0 @@
"""Tests for slack-bot _handle_swap — now creates a pending swap + DMs the target
instead of reassigning immediately."""
from freezegun import freeze_time
MON = "2026-06-01 12:00:00"
def _setup(seed):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.roster("115", "Bob", slack_user_id="U_BOB")
seed.weekly("Monday", "114", "Alice") # Alice is on the Monday shift
@freeze_time(MON)
def test_request_creates_pending_and_dms_target(
slackbot_app, schedule, seed, respond, client, routing_spy, text_of
):
_setup(seed)
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today <@U_BOB>", "C1", client, "C_TEST"
)
swap = schedule.get_swap("2026-06-01")
assert swap["status"] == "pending" and swap["target_ext"] == "115"
# Not applied yet — the shift still belongs to Alice.
assert schedule.get_override("2026-06-01") is None
routing_spy.assert_not_called()
# DM'd the target, told the requester.
assert client.chat_postMessage.call_args.kwargs["channel"] == "U_BOB"
assert "Swap request sent" in text_of(respond)
@freeze_time(MON)
def test_request_by_extension(
slackbot_app, schedule, seed, respond, client, routing_spy
):
_setup(seed)
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today 115", "C1", client, "C_TEST"
)
assert schedule.get_swap("2026-06-01")["target_ext"] == "115"
@freeze_time(MON)
def test_later_request_supersedes(slackbot_app, schedule, seed, respond, client):
_setup(seed)
seed.roster("116", "Carol", slack_user_id="U_CAROL")
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today <@U_BOB>", "C1", client, "C_TEST"
)
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today <@U_CAROL>", "C1", client, "C_TEST"
)
assert schedule.get_swap("2026-06-01")["target_ext"] == "116"
@freeze_time(MON)
def test_target_not_linked_to_slack(
slackbot_app, schedule, seed, respond, client, text_of
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.roster("115", "Bob") # registered but never linked Slack
seed.weekly("Monday", "114", "Alice")
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today 115", "C1", client, "C_TEST"
)
assert "isn't linked to Slack" in text_of(respond)
assert schedule.get_swap("2026-06-01") is None
@freeze_time(MON)
def test_swap_to_self_rejected(slackbot_app, schedule, seed, respond, client, text_of):
_setup(seed)
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today 114", "C1", client, "C_TEST"
)
assert "already yours" in text_of(respond).lower()
@freeze_time(MON)
def test_not_your_shift(slackbot_app, schedule, seed, respond, client, text_of):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.roster("115", "Bob", slack_user_id="U_BOB")
seed.weekly("Monday", "115", "Bob") # Bob's shift, not Alice's
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today <@U_BOB>", "C1", client, "C_TEST"
)
assert "not your shift" in text_of(respond).lower()
assert schedule.get_swap("2026-06-01") is None
@freeze_time(MON)
def test_target_mention_unregistered(
slackbot_app, schedule, seed, respond, client, text_of
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Monday", "114", "Alice")
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today <@U_GHOST>", "C1", client, "C_TEST"
)
assert "isn't registered" in text_of(respond)
@freeze_time(MON)
def test_target_extension_not_found(
slackbot_app, schedule, seed, respond, client, text_of
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.weekly("Monday", "114", "Alice")
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today 999", "C1", client, "C_TEST"
)
assert "not found" in text_of(respond).lower()
@freeze_time(MON)
def test_past_date(slackbot_app, schedule, seed, respond, client, text_of):
_setup(seed)
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap 2026-05-01 <@U_BOB>", "C1", client, "C_TEST"
)
assert "past" in text_of(respond).lower()
@freeze_time(MON)
def test_unregistered_requester(slackbot_app, schedule, respond, client, text_of):
slackbot_app._handle_swap(
respond, schedule, "U_NOBODY", "swap today 115", "C1", client, "C_TEST"
)
assert "not registered" in text_of(respond).lower()
@freeze_time(MON)
def test_usage(slackbot_app, schedule, respond, client, text_of):
slackbot_app._handle_swap(
respond, schedule, "U_ALICE", "swap today", "C1", client, "C_TEST"
)
assert "Usage" in text_of(respond)

View file

@ -1,91 +0,0 @@
"""Tests for slack-bot module-level helpers."""
from freezegun import freeze_time
# Monday 2026-06-01 08:00 ET → weekday, active shift is night.
MON = "2026-06-01 12:00:00"
SAT_DATE = "2026-06-06"
class TestShiftTiming:
def test_shift_start_weekday_night_is_5pm(self, slackbot_app):
assert slackbot_app._shift_start("2026-06-03", "night").hour == 17
def test_shift_start_weekend_day_is_8am(self, slackbot_app):
assert slackbot_app._shift_start("2026-06-06", "day").hour == 8
@freeze_time("2026-06-01 22:00:00") # 18:00 ET
def test_shift_started_true_after_start(self, slackbot_app):
assert slackbot_app._shift_started("2026-06-01", "night") is True
@freeze_time(MON) # 08:00 ET
def test_shift_started_false_before_start(self, slackbot_app):
assert slackbot_app._shift_started("2026-06-01", "night") is False
@freeze_time(MON)
def test_within_drop_lock_for_today(self, slackbot_app):
assert slackbot_app._within_drop_lock("2026-06-01", "night") is True
@freeze_time(MON) # tomorrow 17:00 is ~33h away
def test_not_within_lock_tomorrow_morning(self, slackbot_app):
assert slackbot_app._within_drop_lock("2026-06-02", "night") is False
@freeze_time("2026-06-01 22:00:00") # 18:00 ET, tomorrow 17:00 now <24h
def test_within_lock_once_under_24h(self, slackbot_app):
assert slackbot_app._within_drop_lock("2026-06-02", "night") is True
class TestIsToday:
@freeze_time(MON)
def test_true_for_today(self, slackbot_app):
assert slackbot_app.is_today("2026-06-01") is True
@freeze_time(MON)
def test_false_for_other_day(self, slackbot_app):
assert slackbot_app.is_today("2026-06-02") is False
class TestIsActiveShiftType:
@freeze_time(MON)
def test_weekday_active_is_night(self, slackbot_app):
assert slackbot_app._is_active_shift_type("night") is True
assert slackbot_app._is_active_shift_type("day") is False
class TestShiftTypeLabel:
def test_weekday_has_no_label(self, slackbot_app):
assert slackbot_app._shift_type_label("Wednesday", "night") == ""
def test_weekend_day(self, slackbot_app):
assert slackbot_app._shift_type_label("Saturday", "day") == " (Day)"
def test_weekend_night(self, slackbot_app):
assert slackbot_app._shift_type_label("Saturday", "night") == " (Night)"
class TestFindEmployeeShift:
def test_weekday_night_match(self, slackbot_app, schedule, seed):
seed.weekly("Wednesday", "114", "Alice")
found = slackbot_app._find_employee_shift(
schedule, "2026-06-03", "Wednesday", "114"
)
assert found == ("114", "Alice", "weekly", "night")
def test_weekend_day_match(self, slackbot_app, schedule, seed):
seed.weekly("Saturday", "200", "Alice", shift_type="day")
found = slackbot_app._find_employee_shift(schedule, SAT_DATE, "Saturday", "200")
assert found == ("200", "Alice", "weekly", "day")
def test_weekend_night_match(self, slackbot_app, schedule, seed):
seed.weekly("Saturday", "201", "Bob", shift_type="night")
found = slackbot_app._find_employee_shift(schedule, SAT_DATE, "Saturday", "201")
assert found == ("201", "Bob", "weekly", "night")
def test_not_assigned_returns_none(self, slackbot_app, schedule, seed):
seed.weekly("Wednesday", "114", "Alice")
assert (
slackbot_app._find_employee_shift(
schedule, "2026-06-03", "Wednesday", "999"
)
is None
)

View file

@ -1,59 +0,0 @@
"""Tests for slack-bot parse_date — flexible date parsing."""
from freezegun import freeze_time
# All frozen to Monday 2026-06-01 (ET 08:00).
FROZEN = "2026-06-01 12:00:00"
@freeze_time(FROZEN)
def test_today(slackbot_app):
assert slackbot_app.parse_date("today").strftime("%Y-%m-%d") == "2026-06-01"
@freeze_time(FROZEN)
def test_tomorrow(slackbot_app):
assert slackbot_app.parse_date("tomorrow").strftime("%Y-%m-%d") == "2026-06-02"
@freeze_time(FROZEN)
def test_future_day_name(slackbot_app):
# Friday of the same week.
assert slackbot_app.parse_date("friday").strftime("%Y-%m-%d") == "2026-06-05"
@freeze_time(FROZEN)
def test_same_day_name_jumps_a_week(slackbot_app):
# Monday on a Monday → next Monday, never today.
assert slackbot_app.parse_date("monday").strftime("%Y-%m-%d") == "2026-06-08"
@freeze_time(FROZEN)
def test_iso_date(slackbot_app):
assert slackbot_app.parse_date("2026-07-04").strftime("%Y-%m-%d") == "2026-07-04"
@freeze_time(FROZEN)
def test_month_slash_day(slackbot_app):
assert slackbot_app.parse_date("7/4").strftime("%Y-%m-%d") == "2026-07-04"
@freeze_time(FROZEN)
def test_month_dash_day(slackbot_app):
assert slackbot_app.parse_date("7-4").strftime("%Y-%m-%d") == "2026-07-04"
@freeze_time(FROZEN)
def test_past_month_day_wraps_to_next_year(slackbot_app):
# Jan 1 is well in the past → assume next year.
assert slackbot_app.parse_date("1/1").year == 2027
@freeze_time(FROZEN)
def test_case_insensitive(slackbot_app):
assert slackbot_app.parse_date(" FRIDAY ").strftime("%Y-%m-%d") == "2026-06-05"
@freeze_time(FROZEN)
def test_invalid_returns_none(slackbot_app):
assert slackbot_app.parse_date("notaday") is None

View file

@ -1,73 +0,0 @@
"""Tests for slack-bot handle_pickup (interactive button)."""
from freezegun import freeze_time
MON = "2026-06-01 12:00:00"
def _body(action_id, user_id="U_ALICE", channel="C1"):
return {
"actions": [{"action_id": action_id}],
"user": {"id": user_id},
"channel": {"id": channel},
}
@freeze_time(MON)
def test_pickup_today_success(
slackbot_app, schedule, seed, respond, client, routing_spy
):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
slackbot_app.handle_pickup(
_body("pickup_2026-06-01"), respond, client, schedule, None
)
assert schedule.get_override("2026-06-01")["extension"] == "114"
# Posted to channel and repointed (today + active night).
assert respond.call_args.kwargs.get("response_type") == "in_channel"
routing_spy.assert_called_once_with("114")
@freeze_time(MON)
def test_pickup_day_shift_action_id(slackbot_app, schedule, seed, respond, client):
seed.roster("200", "Alice", slack_user_id="U_ALICE")
slackbot_app.handle_pickup(
_body("pickup_2026-06-06_day"), respond, client, schedule, None
)
# Saturday day-shift override is stored under the -DAY SK.
assert schedule.get_override("2026-06-06", "day")["extension"] == "200"
@freeze_time(MON)
def test_pickup_past_shift_rejected(slackbot_app, schedule, seed, respond, client):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
slackbot_app.handle_pickup(
_body("pickup_2026-05-01"), respond, client, schedule, None
)
client.chat_postEphemeral.assert_called_once()
assert "already passed" in client.chat_postEphemeral.call_args.kwargs["text"]
assert schedule.get_override("2026-05-01") is None
@freeze_time(MON)
def test_pickup_unregistered(slackbot_app, schedule, respond, client):
slackbot_app.handle_pickup(
_body("pickup_2026-06-03", user_id="U_NOBODY"), respond, client, schedule, None
)
assert (
"not registered" in client.chat_postEphemeral.call_args.kwargs["text"].lower()
)
@freeze_time(MON)
def test_pickup_already_taken(slackbot_app, schedule, seed, respond, client):
seed.roster("114", "Alice", slack_user_id="U_ALICE")
seed.override("2026-06-03", "115", "Bob") # already claimed
slackbot_app.handle_pickup(
_body("pickup_2026-06-03"), respond, client, schedule, None
)
assert (
"already picked up"
in client.chat_postEphemeral.call_args.kwargs["text"].lower()
)
# Bob keeps the shift.
assert schedule.get_override("2026-06-03")["extension"] == "115"

View file

@ -1,162 +0,0 @@
"""Tests for slack-bot handle_swap_accept / handle_swap_decline."""
from freezegun import freeze_time
# Monday 2026-06-01 08:00 ET — weekday, active shift is night, before 17:00 start.
MON = "2026-06-01 12:00:00"
# Same Monday but 18:00 ET — the night shift has already started.
MON_AFTER_START = "2026-06-01 22:00:00"
REQ = {"extension": "114", "name": "Alice", "slack_user_id": "U_ALICE"}
TGT = {"extension": "115", "name": "Bob", "slack_user_id": "U_BOB"}
def _accept_body(date_str, user_id="U_BOB", suffix=""):
return {
"actions": [{"action_id": f"swap_accept_{date_str}{suffix}"}],
"user": {"id": user_id},
"channel": {"id": "D1"},
}
def _decline_body(date_str, user_id="U_BOB", suffix=""):
return {
"actions": [{"action_id": f"swap_decline_{date_str}{suffix}"}],
"user": {"id": user_id},
"channel": {"id": "D1"},
}
def _pending(schedule, date_str, shift_type="night", target=TGT):
schedule.create_pending_swap(date_str, shift_type, REQ, target, 1_900_000_000)
def _resolved_text(respond):
return respond.call_args.kwargs["blocks"][0]["text"]["text"]
def _channels(client):
return [c.kwargs.get("channel") for c in client.chat_postMessage.call_args_list]
class TestAccept:
@freeze_time(MON)
def test_applies_override_marks_verified_notifies(
self, slackbot_app, schedule, respond, client, routing_spy
):
_pending(schedule, "2026-06-01")
slackbot_app.handle_swap_accept(
_accept_body("2026-06-01"), respond, client, schedule, "C_TEST"
)
assert schedule.get_override("2026-06-01")["extension"] == "115"
assert schedule.get_swap("2026-06-01")["status"] == "verified"
# Today + active night → 3CX repointed to the new holder.
routing_spy.assert_called_once_with("115")
# Channel notification + requester DM both sent.
assert "C_TEST" in _channels(client) and "U_ALICE" in _channels(client)
assert "now covering" in _resolved_text(respond).lower()
@freeze_time(MON)
def test_future_shift_no_3cx(
self, slackbot_app, schedule, respond, client, routing_spy
):
_pending(schedule, "2026-06-03") # Wednesday
slackbot_app.handle_swap_accept(
_accept_body("2026-06-03"), respond, client, schedule, "C_TEST"
)
assert schedule.get_override("2026-06-03")["extension"] == "115"
routing_spy.assert_not_called()
@freeze_time(MON)
def test_wrong_clicker_rejected(
self, slackbot_app, schedule, respond, client, routing_spy
):
_pending(schedule, "2026-06-01")
slackbot_app.handle_swap_accept(
_accept_body("2026-06-01", user_id="U_STRANGER"),
respond,
client,
schedule,
"C_TEST",
)
assert schedule.get_override("2026-06-01") is None
assert "no longer valid" in _resolved_text(respond).lower()
routing_spy.assert_not_called()
@freeze_time(MON)
def test_no_pending_swap(self, slackbot_app, schedule, respond, client):
slackbot_app.handle_swap_accept(
_accept_body("2026-06-01"), respond, client, schedule, "C_TEST"
)
assert "no longer valid" in _resolved_text(respond).lower()
@freeze_time(MON)
def test_already_verified_swap_not_reapplied(
self, slackbot_app, schedule, respond, client
):
_pending(schedule, "2026-06-01")
schedule.mark_swap_verified("2026-06-01")
slackbot_app.handle_swap_accept(
_accept_body("2026-06-01"), respond, client, schedule, "C_TEST"
)
assert "no longer valid" in _resolved_text(respond).lower()
assert schedule.get_override("2026-06-01") is None
@freeze_time(MON_AFTER_START)
def test_expired_after_shift_start(
self, slackbot_app, schedule, respond, client, routing_spy
):
_pending(schedule, "2026-06-01")
slackbot_app.handle_swap_accept(
_accept_body("2026-06-01"), respond, client, schedule, "C_TEST"
)
assert "expired" in _resolved_text(respond).lower()
assert schedule.get_override("2026-06-01") is None
assert schedule.get_swap("2026-06-01") is None # cleared
routing_spy.assert_not_called()
@freeze_time(MON)
def test_weekend_day_suffix(self, slackbot_app, schedule, respond, client):
_pending(schedule, "2026-06-06", shift_type="day")
slackbot_app.handle_swap_accept(
_accept_body("2026-06-06", suffix="_day"),
respond,
client,
schedule,
"C_TEST",
)
assert schedule.get_override("2026-06-06", "day")["extension"] == "115"
assert schedule.get_swap("2026-06-06", "day")["status"] == "verified"
class TestDecline:
@freeze_time(MON)
def test_clears_and_dms_requester(self, slackbot_app, schedule, respond, client):
_pending(schedule, "2026-06-01")
slackbot_app.handle_swap_decline(
_decline_body("2026-06-01"), respond, client, schedule, "C_TEST"
)
assert schedule.get_swap("2026-06-01") is None
assert schedule.get_override("2026-06-01") is None # never applied
assert "declined" in _resolved_text(respond).lower()
assert "U_ALICE" in _channels(client) # requester notified
@freeze_time(MON)
def test_wrong_clicker_rejected(self, slackbot_app, schedule, respond, client):
_pending(schedule, "2026-06-01")
slackbot_app.handle_swap_decline(
_decline_body("2026-06-01", user_id="U_STRANGER"),
respond,
client,
schedule,
"C_TEST",
)
assert "no longer valid" in _resolved_text(respond).lower()
assert schedule.get_swap("2026-06-01") is not None # untouched
@freeze_time(MON)
def test_no_pending_swap(self, slackbot_app, schedule, respond, client):
slackbot_app.handle_swap_decline(
_decline_body("2026-06-01"), respond, client, schedule, "C_TEST"
)
assert "no longer valid" in _resolved_text(respond).lower()

View file

@ -1,22 +0,0 @@
"""Load src/weekly-post/app.py under a unique module name."""
import importlib.util
import pathlib
import sys
import pytest
_ROOT = pathlib.Path(__file__).resolve().parents[2]
def _load(name, relpath):
spec = importlib.util.spec_from_file_location(name, _ROOT / relpath)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
@pytest.fixture
def weeklypost_app():
return _load("weeklypost_app", "src/weekly-post/app.py")

View file

@ -1,76 +0,0 @@
"""Tests for the weekly-post Lambda handler orchestration."""
from unittest.mock import MagicMock
import pytest
from freezegun import freeze_time
# 2026-06-08 is a Monday. Frozen to ET 08:00; handler is invoked with force=True
# to bypass the 7am DST guard except where the guard itself is under test.
MON_0800 = "2026-06-08 12:00:00"
@pytest.fixture
def slack(weeklypost_app, monkeypatch):
"""Fake Slack WebClient; chat_postMessage returns a message ts."""
fake = MagicMock(name="slack")
fake.chat_postMessage.return_value = {"ts": "999.000"}
monkeypatch.setattr(weeklypost_app, "WebClient", MagicMock(return_value=fake))
monkeypatch.setattr(weeklypost_app, "get_secret", lambda _id: "xoxb-test")
return fake
@pytest.fixture
def env(monkeypatch):
monkeypatch.setenv(
"SLACK_BOT_TOKEN_SECRET", "afterhours-shift-manager/slack-bot-token"
)
monkeypatch.setenv("PAY_REPORT_USER", "U_BOSS")
monkeypatch.delenv("PAYROLL_RECIPIENTS", raising=False) # skip SES email
@freeze_time(MON_0800)
def test_posts_schedule_and_saves_post(weeklypost_app, schedule, seed, slack, env):
result = weeklypost_app.handler({"force": True}, None)
assert result["posted"] is True
assert result["message_ts"] == "999.000"
# The new schedule post was persisted for next week's cleanup.
assert schedule.get_schedule_post("C_TEST")["message_ts"] == "999.000"
slack.chat_postMessage.assert_called()
@freeze_time(MON_0800)
def test_calculates_and_dms_pay(weeklypost_app, schedule, seed, slack, env):
# Previous week (Mon 2026-06-01) had Alice on the Monday night shift.
seed.config(shift_rate="50")
seed.weekly("Monday", "114", "Alice")
result = weeklypost_app.handler({"force": True}, None)
assert result["pay_calculated"] is True
# Pay record saved under previous Monday's key.
assert schedule.get_pay_record("2026-06-01") is not None
# A DM went to the configured pay-report user.
dm_calls = [
c
for c in slack.chat_postMessage.call_args_list
if c.kwargs.get("channel") == "U_BOSS"
]
assert dm_calls
@freeze_time(MON_0800)
def test_deletes_previous_schedule_post(weeklypost_app, schedule, seed, slack, env):
seed.schedule_post("C_TEST", "111.111")
weeklypost_app.handler({"force": True}, None)
slack.chat_delete.assert_called_once()
assert slack.chat_delete.call_args.kwargs["ts"] == "111.111"
@freeze_time(MON_0800)
def test_skips_when_not_7am_and_not_forced(weeklypost_app, schedule, slack, env):
# Frozen hour is 08:00 ET, not 07:00 → skip unless forced.
result = weeklypost_app.handler({}, None)
assert result == {"skipped": True}
slack.chat_postMessage.assert_not_called()

View file

@ -1,83 +0,0 @@
"""Tests for weekly-post pay calculation helpers."""
from datetime import datetime
from decimal import Decimal
# 2026-06-01 is a Monday — required so DAY_ORDER lines up with real weekdays.
WEEK_START = datetime(2026, 6, 1)
class TestAddShiftToPay:
def test_assigned_shift_is_added(self, weeklypost_app, schedule, seed):
seed.config(shift_rate="50")
seed.weekly("Monday", "114", "Alice")
breakdown, totals = [], {}
weeklypost_app._add_shift_to_pay(
schedule, breakdown, totals, datetime(2026, 6, 1), "Monday"
)
assert len(breakdown) == 1
assert breakdown[0]["name"] == "Alice"
assert breakdown[0]["rate"] == Decimal("50")
assert totals["Alice"]["shifts"] == 1
assert totals["Alice"]["total"] == Decimal("50")
def test_available_shift_is_skipped(self, weeklypost_app, schedule, seed):
seed.config(shift_rate="50")
breakdown, totals = [], {}
weeklypost_app._add_shift_to_pay(
schedule, breakdown, totals, datetime(2026, 6, 1), "Monday"
)
assert breakdown == [] and totals == {}
def test_weekend_day_label(self, weeklypost_app, schedule, seed):
seed.config(shift_rate="50")
seed.weekly("Saturday", "200", "Alice", shift_type="day")
breakdown, totals = [], {}
weeklypost_app._add_shift_to_pay(
schedule, breakdown, totals, datetime(2026, 6, 6), "Saturday", "day"
)
assert breakdown[0]["day"] == "Sat (Day)"
class TestCalculateWeeklyPay:
def test_single_weekday_shift(self, weeklypost_app, schedule, seed):
seed.config(shift_rate="50")
seed.weekly("Monday", "114", "Alice")
record = weeklypost_app._calculate_weekly_pay(schedule, WEEK_START)
assert record["week_start"] == "2026-06-01"
assert len(record["breakdown"]) == 1
assert record["totals"]["Alice"]["total"] == Decimal("50")
def test_per_employee_rate_applied(self, weeklypost_app, schedule, seed):
seed.config(shift_rate="50")
seed.roster("114", "Alice", shift_rate="75")
seed.weekly("Monday", "114", "Alice")
record = weeklypost_app._calculate_weekly_pay(schedule, WEEK_START)
assert record["totals"]["Alice"]["total"] == Decimal("75.0")
def test_weekend_counts_day_and_night(self, weeklypost_app, schedule, seed):
seed.config(shift_rate="50")
seed.weekly("Saturday", "200", "Alice", shift_type="day")
seed.weekly("Saturday", "200", "Alice", shift_type="night")
record = weeklypost_app._calculate_weekly_pay(schedule, WEEK_START)
# Alice worked both Saturday shifts → 2 shifts, $100.
assert record["totals"]["Alice"]["shifts"] == 2
assert record["totals"]["Alice"]["total"] == Decimal("100")
def test_empty_week_has_no_breakdown(self, weeklypost_app, schedule, seed):
seed.config(shift_rate="50")
record = weeklypost_app._calculate_weekly_pay(schedule, WEEK_START)
assert record["breakdown"] == [] and record["totals"] == {}
class TestBuildPayEmailHtml:
def test_renders_totals_rows(self, weeklypost_app):
pay_record = {
"totals": {
"Alice": {"total": Decimal("100"), "rate": Decimal("50"), "shifts": 2}
}
}
html = weeklypost_app._build_pay_email_html("Jun 1 to Jun 7", pay_record)
assert "Jun 1 to Jun 7" in html
assert "Alice" in html
assert "$100.00" in html