feat: add team calendar ICS endpoint and related functionality
All checks were successful
CI / lint-and-test (push) Successful in 23s

- Implemented a new API endpoint to generate an ICS calendar for team duty shifts, accessible via a valid token.
- Enhanced the `calendar_link` command to return both personal and team calendar URLs.
- Added a new function to build the team ICS file, ensuring each event includes the duty holder's name in the description.
- Updated tests to cover the new team calendar functionality, including validation for token formats and response content.
- Revised internationalization messages to reflect the new team calendar links.
This commit is contained in:
2026-02-21 23:41:00 +03:00
parent 0c93fe3372
commit 77a94fa91b
8 changed files with 183 additions and 17 deletions

BIN
.coverage

Binary file not shown.

View File

@@ -17,8 +17,12 @@ from duty_teller.api.dependencies import (
get_validated_dates, get_validated_dates,
require_miniapp_username, require_miniapp_username,
) )
from duty_teller.api.personal_calendar_ics import build_personal_ics from duty_teller.api.personal_calendar_ics import build_personal_ics, build_team_ics
from duty_teller.db.repository import get_duties_for_user, get_user_by_calendar_token from duty_teller.db.repository import (
get_duties,
get_duties_for_user,
get_user_by_calendar_token,
)
from duty_teller.db.schemas import CalendarEvent, DutyWithUser from duty_teller.db.schemas import CalendarEvent, DutyWithUser
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -94,6 +98,38 @@ def list_calendar_events(
return [CalendarEvent(date=e["date"], summary=e["summary"]) for e in events] return [CalendarEvent(date=e["date"], summary=e["summary"]) for e in events]
@app.get(
"/api/calendar/ical/team/{token}.ics",
summary="Team calendar ICS",
description=(
"Returns an ICS calendar with all team duty shifts. Each event has "
"DESCRIPTION set to the duty holder's name. No Telegram auth; access by token."
),
)
def get_team_calendar_ical(
token: str,
session: Session = Depends(get_db_session),
) -> Response:
"""Return ICS calendar with all duties (event_type duty only). Token validates user."""
if not _is_valid_calendar_token(token):
return Response(status_code=404, content="Not found")
user = get_user_by_calendar_token(session, token)
if user is None:
return Response(status_code=404, content="Not found")
today = date.today()
from_date = (today - timedelta(days=365)).strftime("%Y-%m-%d")
to_date = (today + timedelta(days=365 * 2)).strftime("%Y-%m-%d")
all_duties = get_duties(session, from_date=from_date, to_date=to_date)
duties_duty_only = [
(d, name) for d, name in all_duties if (d.event_type or "duty") == "duty"
]
ics_bytes = build_team_ics(duties_duty_only)
return Response(
content=ics_bytes,
media_type="text/calendar; charset=utf-8",
)
@app.get( @app.get(
"/api/calendar/ical/{token}.ics", "/api/calendar/ical/{token}.ics",
summary="Personal calendar ICS", summary="Personal calendar ICS",

View File

@@ -50,3 +50,39 @@ def build_personal_ics(duties_with_name: list[tuple[Duty, str]]) -> bytes:
cal.add_component(event) cal.add_component(event)
return cal.to_ical() return cal.to_ical()
def build_team_ics(duties_with_name: list[tuple[Duty, str]]) -> bytes:
"""Build a VCALENDAR (ICS) with one VEVENT per duty for team calendar.
Same structure as personal calendar; PRODID is Team Calendar; each VEVENT
has SUMMARY='Duty' and DESCRIPTION set to the duty holder's full_name.
Args:
duties_with_name: List of (Duty, full_name). full_name is used for DESCRIPTION.
Returns:
ICS file content as bytes (UTF-8).
"""
cal = Calendar()
cal.add("prodid", "-//Duty Teller//Team Calendar//EN")
cal.add("version", "2.0")
cal.add("calscale", "GREGORIAN")
for duty, full_name in duties_with_name:
event = Event()
start_dt = parse_utc_iso(duty.start_at)
end_dt = parse_utc_iso(duty.end_at)
if start_dt.tzinfo is None:
start_dt = start_dt.replace(tzinfo=timezone.utc)
if end_dt.tzinfo is None:
end_dt = end_dt.replace(tzinfo=timezone.utc)
event.add("dtstart", start_dt)
event.add("dtend", end_dt)
event.add("summary", "Duty")
event.add("description", full_name)
event.add("uid", f"duty-{duty.id}@duty-teller")
event.add("dtstamp", datetime.now(timezone.utc))
cal.add_component(event)
return cal.to_ical()

View File

@@ -122,21 +122,30 @@ async def calendar_link(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N
if not can_access_miniapp_for_telegram_user(session, telegram_user_id): if not can_access_miniapp_for_telegram_user(session, telegram_user_id):
return (None, "denied") return (None, "denied")
token = create_calendar_token(session, user.id) token = create_calendar_token(session, user.id)
base = (config.MINI_APP_BASE_URL or "").rstrip("/") return (token, None)
url = f"{base}/api/calendar/ical/{token}.ics" if base else None
return (url, None)
result_url, error = await asyncio.get_running_loop().run_in_executor( result_token, error = await asyncio.get_running_loop().run_in_executor(
None, do_calendar_link None, do_calendar_link
) )
if error == "denied": if error == "denied":
await update.message.reply_text(t(lang, "calendar_link.access_denied")) await update.message.reply_text(t(lang, "calendar_link.access_denied"))
return return
if not result_url: if not result_token:
await update.message.reply_text(t(lang, "calendar_link.error")) await update.message.reply_text(t(lang, "calendar_link.error"))
return return
base = (config.MINI_APP_BASE_URL or "").rstrip("/")
if not base:
await update.message.reply_text(t(lang, "calendar_link.error"))
return
url_personal = f"{base}/api/calendar/ical/{result_token}.ics"
url_team = f"{base}/api/calendar/ical/team/{result_token}.ics"
await update.message.reply_text( await update.message.reply_text(
t(lang, "calendar_link.success", url=result_url) t(
lang,
"calendar_link.success",
url_personal=url_personal,
url_team=url_team,
)
+ "\n\n" + "\n\n"
+ t(lang, "calendar_link.help_hint") + t(lang, "calendar_link.help_hint")
) )

View File

@@ -15,10 +15,12 @@ MESSAGES: dict[str, dict[str, str]] = {
"help.pin_duty": "/pin_duty — In a group: pin the duty message (bot needs admin with Pin messages)", "help.pin_duty": "/pin_duty — In a group: pin the duty message (bot needs admin with Pin messages)",
"calendar_link.private_only": "The /calendar_link command is only available in private chat.", "calendar_link.private_only": "The /calendar_link command is only available in private chat.",
"calendar_link.access_denied": "Access denied.", "calendar_link.access_denied": "Access denied.",
"calendar_link.success": "Your personal calendar URL:\n{url}", "calendar_link.success": (
"Personal (your duties only):\n{url_personal}\n\n"
"Team (all duties):\n{url_team}"
),
"calendar_link.help_hint": ( "calendar_link.help_hint": (
"Subscribe to this URL in Google Calendar, Apple Calendar, or Outlook to " "Subscribe to these URLs in Google Calendar, Apple Calendar, or Outlook."
"see only your duties."
), ),
"calendar_link.error": "Could not generate link. Please try again later.", "calendar_link.error": "Could not generate link. Please try again later.",
"help.import_schedule": "/import_duty_schedule — Import duty schedule (JSON)", "help.import_schedule": "/import_duty_schedule — Import duty schedule (JSON)",
@@ -82,8 +84,13 @@ MESSAGES: dict[str, dict[str, str]] = {
"help.pin_duty": "/pin_duty — В группе: закрепить сообщение о дежурстве (нужны права админа у бота)", "help.pin_duty": "/pin_duty — В группе: закрепить сообщение о дежурстве (нужны права админа у бота)",
"calendar_link.private_only": "Команда /calendar_link доступна только в личке.", "calendar_link.private_only": "Команда /calendar_link доступна только в личке.",
"calendar_link.access_denied": "Доступ запрещён.", "calendar_link.access_denied": "Доступ запрещён.",
"calendar_link.success": "Ссылка на ваш календарь:\n{url}", "calendar_link.success": (
"calendar_link.help_hint": "Подпишитесь на эту ссылку в Google Календаре, Календаре Apple или Outlook, чтобы видеть только свои дежурства.", "Персональный (только ваши дежурства):\n{url_personal}\n\n"
"Общий (все дежурства):\n{url_team}"
),
"calendar_link.help_hint": (
"Подпишитесь на эти ссылки в Google Календаре, Календаре Apple или Outlook."
),
"calendar_link.error": "Не удалось сформировать ссылку. Попробуйте позже.", "calendar_link.error": "Не удалось сформировать ссылку. Попробуйте позже.",
"help.import_schedule": "/import_duty_schedule — Импорт расписания дежурств (JSON)", "help.import_schedule": "/import_duty_schedule — Импорт расписания дежурств (JSON)",
"help.set_role": "/set_role — Выдать роль пользователю (user | admin)", "help.set_role": "/set_role — Выдать роль пользователю (user | admin)",

View File

@@ -268,6 +268,67 @@ def test_duties_200_with_unknown_event_type_mapped_to_duty(client):
assert data[0]["full_name"] == "User A" assert data[0]["full_name"] == "User A"
def test_calendar_ical_team_404_invalid_token_format(client):
"""GET /api/calendar/ical/team/{token}.ics with invalid token format returns 404 without DB."""
r = client.get("/api/calendar/ical/team/short.ics")
assert r.status_code == 404
assert "not found" in r.text.lower()
@patch("duty_teller.api.app.get_user_by_calendar_token")
def test_calendar_ical_team_404_unknown_token(mock_get_user, client):
"""GET /api/calendar/ical/team/{token}.ics with unknown token returns 404."""
mock_get_user.return_value = None
valid_format_token = "B" * 43
r = client.get(f"/api/calendar/ical/team/{valid_format_token}.ics")
assert r.status_code == 404
assert "not found" in r.text.lower()
mock_get_user.assert_called_once()
@patch("duty_teller.api.app.build_team_ics")
@patch("duty_teller.api.app.get_duties")
@patch("duty_teller.api.app.get_user_by_calendar_token")
def test_calendar_ical_team_200_only_duty_and_description(
mock_get_user, mock_get_duties, mock_build_team_ics, client
):
"""GET /api/calendar/ical/team/{token}.ics returns ICS with duty-only events and DESCRIPTION."""
from types import SimpleNamespace
mock_user = SimpleNamespace(id=1, full_name="User A")
mock_get_user.return_value = mock_user
duty = SimpleNamespace(
id=10,
user_id=1,
start_at="2026-06-15T09:00:00Z",
end_at="2026-06-15T18:00:00Z",
event_type="duty",
)
non_duty = SimpleNamespace(
id=11,
user_id=2,
start_at="2026-06-16T09:00:00Z",
end_at="2026-06-16T18:00:00Z",
event_type="vacation",
)
mock_get_duties.return_value = [(duty, "User A"), (non_duty, "User B")]
mock_build_team_ics.return_value = b"BEGIN:VCALENDAR\r\nPRODID:Team\r\nVEVENT\r\nDESCRIPTION:User A\r\nEND:VCALENDAR"
token = "y" * 43
r = client.get(f"/api/calendar/ical/team/{token}.ics")
assert r.status_code == 200
assert r.headers.get("content-type", "").startswith("text/calendar")
assert b"BEGIN:VCALENDAR" in r.content
mock_get_user.assert_called_once()
mock_get_duties.assert_called_once()
# build_team_ics called with only duty (event_type duty), not vacation
mock_build_team_ics.assert_called_once()
duties_arg = mock_build_team_ics.call_args[0][0]
assert len(duties_arg) == 1
assert duties_arg[0][0].event_type == "duty"
assert duties_arg[0][1] == "User A"
def test_calendar_ical_404_invalid_token_format(client): def test_calendar_ical_404_invalid_token_format(client):
"""GET /api/calendar/ical/{token}.ics with invalid token format returns 404 without DB call.""" """GET /api/calendar/ical/{token}.ics with invalid token format returns 404 without DB call."""
# Token format must be base64url, 4050 chars; short or invalid chars → 404 # Token format must be base64url, 4050 chars; short or invalid chars → 404

View File

@@ -202,7 +202,7 @@ async def test_set_phone_result_error_replies_error():
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_calendar_link_with_user_and_token_replies_with_url(): async def test_calendar_link_with_user_and_token_replies_with_url():
"""calendar_link with allowed user and token -> reply with link.""" """calendar_link with allowed user and token -> reply with personal and team URLs."""
message = MagicMock() message = MagicMock()
message.reply_text = AsyncMock() message.reply_text = AsyncMock()
user = _make_user() user = _make_user()
@@ -233,14 +233,15 @@ async def test_calendar_link_with_user_and_token_replies_with_url():
): ):
with patch("duty_teller.handlers.commands.t") as mock_t: with patch("duty_teller.handlers.commands.t") as mock_t:
mock_t.side_effect = lambda lang, key, **kw: ( mock_t.side_effect = lambda lang, key, **kw: (
f"URL: {kw.get('url', '')}" f"Personal: {kw.get('url_personal', '')} Team: {kw.get('url_team', '')}"
if "success" in key if "success" in key
else "Hint" else "Hint"
) )
await calendar_link(update, MagicMock()) await calendar_link(update, MagicMock())
message.reply_text.assert_called_once() message.reply_text.assert_called_once()
call_args = message.reply_text.call_args[0][0] call_args = message.reply_text.call_args[0][0]
assert "abc43token" in call_args or "example.com" in call_args assert "/api/calendar/ical/abc43token.ics" in call_args
assert "/api/calendar/ical/team/abc43token.ics" in call_args
@pytest.mark.asyncio @pytest.mark.asyncio

View File

@@ -4,7 +4,7 @@ from types import SimpleNamespace
from icalendar import Calendar as ICalendar from icalendar import Calendar as ICalendar
from duty_teller.api.personal_calendar_ics import build_personal_ics from duty_teller.api.personal_calendar_ics import build_personal_ics, build_team_ics
# Minimal Duty-like object for build_personal_ics(duties_with_name: list[tuple[Duty, str]]) # Minimal Duty-like object for build_personal_ics(duties_with_name: list[tuple[Duty, str]])
@@ -45,6 +45,22 @@ def test_build_personal_ics_one_duty():
assert b"duty-1@duty-teller" in ics or "duty-1@duty-teller" in ics.decode("utf-8") assert b"duty-1@duty-teller" in ics or "duty-1@duty-teller" in ics.decode("utf-8")
def test_build_team_ics_vevent_has_description_with_full_name():
"""Team ICS VEVENT includes DESCRIPTION with the duty holder's full_name."""
duties = [
_duty(1, "2025-02-20T09:00:00Z", "2025-02-20T18:00:00Z", "duty"),
]
ics = build_team_ics(duties)
cal = ICalendar.from_ical(ics)
assert cal is not None
events = [c for c in cal.walk() if c.name == "VEVENT"]
assert len(events) == 1
ev = events[0]
assert ev.get("summary") == "Duty"
assert ev.get("description") == "Test User"
assert b"Team Calendar" in ics or "Team Calendar" in ics.decode("utf-8")
def test_build_personal_ics_event_types(): def test_build_personal_ics_event_types():
"""Unavailable and vacation get correct SUMMARY.""" """Unavailable and vacation get correct SUMMARY."""
duties = [ duties = [