Nikolay Tatarinov
87e8417675
- Refactored the `mini_app_short_name` assignment in `config.py` for better clarity. - Enhanced the `app_config_js` function in `app.py` to improve formatting of the JavaScript response body. - Added per-chat locks in `group_duty_pin.py` to prevent concurrent refreshes, improving message handling. - Updated `_schedule_next_update` to include optional jitter for scheduling, enhancing performance during high-load scenarios. - Cleaned up test files by removing unused imports and improving test descriptions for clarity.
590 lines
22 KiB
Python
590 lines
22 KiB
Python
"""Tests for FastAPI app /api/duties."""
|
||
|
||
import time
|
||
from unittest.mock import ANY, patch
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
import duty_teller.config as config
|
||
from duty_teller.api.app import app
|
||
from tests.helpers import make_init_data
|
||
|
||
|
||
@pytest.fixture
|
||
def client():
|
||
return TestClient(app)
|
||
|
||
|
||
def test_health(client):
|
||
"""Health endpoint returns 200 and status ok for Docker HEALTHCHECK."""
|
||
r = client.get("/health")
|
||
assert r.status_code == 200
|
||
assert r.json() == {"status": "ok"}
|
||
|
||
|
||
def test_unhandled_exception_returns_500_json(client):
|
||
"""Global exception handler returns 500 JSON without leaking exception details."""
|
||
from unittest.mock import MagicMock
|
||
|
||
from duty_teller.api.app import global_exception_handler
|
||
|
||
# Call the registered handler directly: it returns JSON and does not expose str(exc).
|
||
request = MagicMock()
|
||
exc = RuntimeError("internal failure")
|
||
response = global_exception_handler(request, exc)
|
||
assert response.status_code == 500
|
||
assert response.body.decode() == '{"detail":"Internal server error"}'
|
||
assert "internal failure" not in response.body.decode()
|
||
|
||
|
||
def test_health_has_vary_accept_language(client):
|
||
"""NoCacheStaticMiddleware adds Vary: Accept-Language to all responses."""
|
||
r = client.get("/health")
|
||
assert r.status_code == 200
|
||
assert "accept-language" in r.headers.get("vary", "").lower()
|
||
|
||
|
||
def test_app_static_has_no_store_and_vary(client):
|
||
"""Static files under /app get Cache-Control: no-store and Vary: Accept-Language."""
|
||
r = client.get("/app/")
|
||
if r.status_code != 200:
|
||
r = client.get("/app")
|
||
assert r.status_code == 200, (
|
||
"webapp static mount should serve index at /app or /app/"
|
||
)
|
||
assert r.headers.get("cache-control") == "no-store"
|
||
assert "accept-language" in r.headers.get("vary", "").lower()
|
||
|
||
|
||
def test_app_js_has_no_store(client):
|
||
"""JS and all static under /app get Cache-Control: no-store."""
|
||
webapp_out = config.PROJECT_ROOT / "webapp-next" / "out"
|
||
if not webapp_out.is_dir():
|
||
pytest.skip("webapp-next/out not built")
|
||
# Next.js static export serves JS under _next/static/chunks/<hash>.js
|
||
js_files = list(webapp_out.glob("_next/static/chunks/*.js"))
|
||
if not js_files:
|
||
pytest.skip("no JS chunks in webapp-next/out")
|
||
rel = js_files[0].relative_to(webapp_out)
|
||
r = client.get(f"/app/{rel.as_posix()}")
|
||
assert r.status_code == 200
|
||
assert r.headers.get("cache-control") == "no-store"
|
||
|
||
|
||
def test_app_config_js_returns_lang_from_default_language(client):
|
||
"""GET /app/config.js returns JS setting window.__DT_LANG from config.DEFAULT_LANGUAGE."""
|
||
r = client.get("/app/config.js")
|
||
assert r.status_code == 200
|
||
assert r.headers.get("content-type", "").startswith("application/javascript")
|
||
assert r.headers.get("cache-control") == "no-store"
|
||
body = r.text
|
||
assert "window.__DT_LANG" in body
|
||
assert config.DEFAULT_LANGUAGE in body
|
||
|
||
|
||
@patch("duty_teller.api.app.config.DEFAULT_LANGUAGE", '"; alert(1); "')
|
||
@patch("duty_teller.api.app.config.LOG_LEVEL_STR", "DEBUG\x00INJECT")
|
||
def test_app_config_js_sanitizes_lang_and_log_level(client):
|
||
"""config.js uses whitelist: invalid lang/log_level produce safe defaults, no script injection."""
|
||
r = client.get("/app/config.js")
|
||
assert r.status_code == 200
|
||
body = r.text
|
||
# Must be valid JS and not contain the raw malicious strings.
|
||
assert 'window.__DT_LANG = "en"' in body or 'window.__DT_LANG = "ru"' in body
|
||
assert "alert" not in body
|
||
assert "INJECT" not in body
|
||
assert "window.__DT_LOG_LEVEL" in body
|
||
|
||
|
||
def test_duties_invalid_date_format(client):
|
||
r = client.get("/api/duties", params={"from": "2025-01-01", "to": "invalid"})
|
||
assert r.status_code == 400
|
||
detail = r.json()["detail"]
|
||
assert "from" in detail.lower() and "to" in detail.lower()
|
||
|
||
|
||
def test_duties_from_after_to(client):
|
||
r = client.get("/api/duties", params={"from": "2025-02-01", "to": "2025-01-01"})
|
||
assert r.status_code == 400
|
||
detail = r.json()["detail"].lower()
|
||
assert "from" in detail or "to" in detail or "after" in detail or "позже" in detail
|
||
|
||
|
||
@patch("duty_teller.api.app.config.MINI_APP_SKIP_AUTH", True)
|
||
def test_duties_range_too_large_400(client):
|
||
"""Date range longer than MAX_DATE_RANGE_DAYS returns 400 with dates.range_too_large message."""
|
||
from datetime import date, timedelta
|
||
|
||
from duty_teller.utils.dates import MAX_DATE_RANGE_DAYS
|
||
|
||
from_d = date(2020, 1, 1)
|
||
to_d = from_d + timedelta(days=MAX_DATE_RANGE_DAYS + 1)
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": from_d.isoformat(), "to": to_d.isoformat()},
|
||
)
|
||
assert r.status_code == 400
|
||
detail = r.json()["detail"]
|
||
# EN: "Date range is too large. Request a shorter period." / RU: "Диапазон дат слишком большой..."
|
||
assert (
|
||
"range" in detail.lower()
|
||
or "short" in detail.lower()
|
||
or "короткий" in detail
|
||
or "большой" in detail
|
||
)
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.config.MINI_APP_SKIP_AUTH", False)
|
||
def test_duties_403_without_init_data(client):
|
||
"""Without X-Telegram-Init-Data and without MINI_APP_SKIP_AUTH → 403 (any client)."""
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
)
|
||
assert r.status_code == 403
|
||
|
||
|
||
@patch("duty_teller.api.app.config.MINI_APP_SKIP_AUTH", True)
|
||
@patch("duty_teller.api.app.fetch_duties_response")
|
||
def test_duties_200_when_skip_auth(mock_fetch, client):
|
||
mock_fetch.return_value = []
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json() == []
|
||
mock_fetch.assert_called_once_with(ANY, "2025-01-01", "2025-01-31")
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
|
||
@patch("duty_teller.api.dependencies.config.MINI_APP_SKIP_AUTH", False)
|
||
def test_duties_403_when_init_data_invalid(mock_validate, client):
|
||
mock_validate.return_value = (None, None, "hash_mismatch", "en")
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
headers={"X-Telegram-Init-Data": "some=data&hash=abc"},
|
||
)
|
||
assert r.status_code == 403
|
||
detail = r.json()["detail"]
|
||
assert (
|
||
"signature" in detail.lower()
|
||
or "авторизации" in detail
|
||
or "Неверные" in detail
|
||
or "Неверная" in detail
|
||
)
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.get_user_by_telegram_id")
|
||
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
|
||
@patch("duty_teller.api.dependencies.config.MINI_APP_SKIP_AUTH", False)
|
||
def test_duties_403_when_user_not_in_db(mock_validate, mock_get_user, client):
|
||
"""User not in DB -> 403 (access only for users with role or env-admin fallback)."""
|
||
mock_validate.return_value = (123, "someuser", "ok", "en")
|
||
mock_get_user.return_value = None
|
||
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
headers={"X-Telegram-Init-Data": "user=xxx&hash=yyy"},
|
||
)
|
||
assert r.status_code == 403
|
||
assert (
|
||
"Access denied" in r.json()["detail"] or "Доступ запрещён" in r.json()["detail"]
|
||
)
|
||
mock_fetch.assert_not_called()
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.can_access_miniapp_for_telegram_user")
|
||
@patch("duty_teller.api.dependencies.get_user_by_telegram_id")
|
||
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
|
||
@patch("duty_teller.api.dependencies.config.MINI_APP_SKIP_AUTH", False)
|
||
def test_duties_403_when_no_username_and_no_access(
|
||
mock_validate, mock_get_user, mock_can_access, client
|
||
):
|
||
"""User in DB but no miniapp access (no role, not env admin) -> 403."""
|
||
from types import SimpleNamespace
|
||
|
||
mock_validate.return_value = (456, None, "ok", "en")
|
||
mock_get_user.return_value = SimpleNamespace(
|
||
phone="+79001111111", full_name="User", username=None
|
||
)
|
||
mock_can_access.return_value = False
|
||
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
headers={"X-Telegram-Init-Data": "user=xxx&hash=yyy"},
|
||
)
|
||
assert r.status_code == 403
|
||
mock_fetch.assert_not_called()
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.can_access_miniapp_for_telegram_user")
|
||
@patch("duty_teller.api.dependencies.get_user_by_telegram_id")
|
||
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
|
||
@patch("duty_teller.api.dependencies.config.MINI_APP_SKIP_AUTH", False)
|
||
def test_duties_200_when_user_has_access(
|
||
mock_validate, mock_get_user, mock_can_access, client
|
||
):
|
||
"""User in DB with miniapp access (role or env fallback) -> 200."""
|
||
from types import SimpleNamespace
|
||
|
||
mock_validate.return_value = (789, None, "ok", "en")
|
||
mock_get_user.return_value = SimpleNamespace(
|
||
phone="+7 900 123-45-67", full_name="Иван", username=None
|
||
)
|
||
mock_can_access.return_value = True
|
||
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
|
||
mock_fetch.return_value = []
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
headers={"X-Telegram-Init-Data": "user=xxx&hash=yyy"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json() == []
|
||
mock_fetch.assert_called_once_with(ANY, "2025-01-01", "2025-01-31")
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
|
||
@patch("duty_teller.api.dependencies.config.can_access_miniapp")
|
||
@patch("duty_teller.api.dependencies.config.MINI_APP_SKIP_AUTH", True)
|
||
def test_duties_200_with_valid_init_data_and_skip_auth_not_in_allowlist(
|
||
mock_can_access, mock_validate, client
|
||
):
|
||
"""With MINI_APP_SKIP_AUTH=True, valid initData grants access without allowlist check."""
|
||
mock_validate.return_value = ("outsider_user", "ok", "en")
|
||
mock_can_access.return_value = False
|
||
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
|
||
mock_fetch.return_value = []
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
headers={"X-Telegram-Init-Data": "user=xxx&hash=yyy"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json() == []
|
||
mock_fetch.assert_called_once_with(ANY, "2025-01-01", "2025-01-31")
|
||
mock_can_access.assert_not_called()
|
||
mock_validate.assert_not_called()
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.can_access_miniapp_for_telegram_user")
|
||
@patch("duty_teller.api.dependencies.get_user_by_telegram_id")
|
||
@patch("duty_teller.api.dependencies.validate_init_data_with_reason")
|
||
def test_duties_200_with_allowed_user(
|
||
mock_validate, mock_get_user, mock_can_access, client
|
||
):
|
||
from types import SimpleNamespace
|
||
|
||
mock_validate.return_value = (1, "alloweduser", "ok", "en")
|
||
mock_get_user.return_value = SimpleNamespace(
|
||
full_name="Иван Иванов", username="alloweduser", phone=None
|
||
)
|
||
mock_can_access.return_value = True
|
||
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
|
||
mock_fetch.return_value = [
|
||
{
|
||
"id": 1,
|
||
"user_id": 10,
|
||
"start_at": "2025-01-15T09:00:00Z",
|
||
"end_at": "2025-01-15T18:00:00Z",
|
||
"full_name": "Иван Иванов",
|
||
}
|
||
]
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
headers={"X-Telegram-Init-Data": "user=xxx&hash=yyy"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert len(r.json()) == 1
|
||
assert r.json()[0]["full_name"] == "Иван Иванов"
|
||
mock_fetch.assert_called_once_with(ANY, "2025-01-01", "2025-01-31")
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.can_access_miniapp_for_telegram_user")
|
||
@patch("duty_teller.api.dependencies.get_user_by_telegram_id")
|
||
def test_duties_e2e_auth_real_validation(
|
||
mock_get_user, mock_can_access, client, monkeypatch
|
||
):
|
||
from types import SimpleNamespace
|
||
|
||
test_token = "123:ABC"
|
||
test_username = "e2euser"
|
||
monkeypatch.setattr(config, "BOT_TOKEN", test_token)
|
||
monkeypatch.setattr(config, "ADMIN_USERNAMES", set())
|
||
monkeypatch.setattr(config, "INIT_DATA_MAX_AGE_SECONDS", 0)
|
||
init_data = make_init_data(
|
||
{"id": 1, "username": test_username},
|
||
test_token,
|
||
auth_date=int(time.time()),
|
||
)
|
||
mock_get_user.return_value = SimpleNamespace(
|
||
full_name="E2E User", username=test_username, phone=None
|
||
)
|
||
mock_can_access.return_value = True
|
||
with patch("duty_teller.api.app.fetch_duties_response") as mock_fetch:
|
||
mock_fetch.return_value = []
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
headers={"X-Telegram-Init-Data": init_data},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json() == []
|
||
mock_fetch.assert_called_once_with(ANY, "2025-01-01", "2025-01-31")
|
||
|
||
|
||
@patch("duty_teller.api.app.config.MINI_APP_SKIP_AUTH", True)
|
||
def test_duties_200_with_unknown_event_type_mapped_to_duty(client):
|
||
from types import SimpleNamespace
|
||
|
||
fake_duty = SimpleNamespace(
|
||
id=1,
|
||
user_id=10,
|
||
start_at="2025-01-15T09:00:00Z",
|
||
end_at="2025-01-15T18:00:00Z",
|
||
event_type="unknown",
|
||
)
|
||
|
||
def fake_get_duties(session, from_date, to_date):
|
||
# get_duties returns (Duty, full_name, phone, username) tuples.
|
||
return [(fake_duty, "User A", "+79001234567", "user_a")]
|
||
|
||
with patch("duty_teller.api.dependencies.get_duties", side_effect=fake_get_duties):
|
||
r = client.get(
|
||
"/api/duties",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
)
|
||
assert r.status_code == 200
|
||
data = r.json()
|
||
assert len(data) == 1
|
||
assert data[0]["event_type"] == "duty"
|
||
assert data[0]["full_name"] == "User A"
|
||
assert data[0].get("phone") == "+79001234567"
|
||
assert data[0].get("username") == "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 JSON."""
|
||
r = client.get("/api/calendar/ical/team/short.ics")
|
||
assert r.status_code == 404
|
||
assert r.headers.get("content-type", "").startswith("application/json")
|
||
assert r.json() == {"detail": "Not found"}
|
||
|
||
|
||
@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 JSON."""
|
||
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 r.json() == {"detail": "Not found"}
|
||
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",
|
||
)
|
||
# get_duties returns (Duty, full_name, phone, username) tuples.
|
||
mock_get_duties.return_value = [
|
||
(duty, "User A", None, None),
|
||
(non_duty, "User B", None, None),
|
||
]
|
||
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):
|
||
"""GET /api/calendar/ical/{token}.ics with invalid token format returns 404 JSON."""
|
||
r = client.get("/api/calendar/ical/short.ics")
|
||
assert r.status_code == 404
|
||
assert r.json() == {"detail": "Not found"}
|
||
r2 = client.get("/api/calendar/ical/" + "x" * 60 + ".ics")
|
||
assert r2.status_code == 404
|
||
r3 = client.get("/api/calendar/ical/../../../etc/passwd.ics")
|
||
assert r3.status_code == 404
|
||
|
||
|
||
@patch("duty_teller.api.app.get_user_by_calendar_token")
|
||
def test_calendar_ical_404_unknown_token(mock_get_user, client):
|
||
"""GET /api/calendar/ical/{token}.ics with unknown token returns 404 JSON."""
|
||
mock_get_user.return_value = None
|
||
valid_format_token = "A" * 43
|
||
r = client.get(f"/api/calendar/ical/{valid_format_token}.ics")
|
||
assert r.status_code == 404
|
||
assert r.json() == {"detail": "Not found"}
|
||
mock_get_user.assert_called_once()
|
||
|
||
|
||
@patch("duty_teller.api.app.build_personal_ics")
|
||
@patch("duty_teller.api.app.get_duties_for_user")
|
||
@patch("duty_teller.api.app.get_user_by_calendar_token")
|
||
def test_calendar_ical_200_returns_only_that_users_duties(
|
||
mock_get_user, mock_get_duties, mock_build_ics, client
|
||
):
|
||
"""GET /api/calendar/ical/{token}.ics returns ICS with only the token owner's duties."""
|
||
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",
|
||
)
|
||
# get_duties_for_user returns (Duty, full_name, phone, username) tuples.
|
||
mock_get_duties.return_value = [(duty, "User A", None, None)]
|
||
mock_build_ics.return_value = (
|
||
b"BEGIN:VCALENDAR\r\nVEVENT\r\n2026-06-15\r\nEND:VCALENDAR"
|
||
)
|
||
# Token must pass format validation (base64url, 40–50 chars)
|
||
token = "x" * 43
|
||
|
||
r = client.get(f"/api/calendar/ical/{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_with(
|
||
ANY, 1, from_date=ANY, to_date=ANY, event_types=["duty"]
|
||
)
|
||
mock_build_ics.assert_called_once()
|
||
# Only User A's duty was passed to build_personal_ics
|
||
duties_arg = mock_build_ics.call_args[0][0]
|
||
assert len(duties_arg) == 1
|
||
assert duties_arg[0][0].user_id == 1
|
||
assert duties_arg[0][1] == "User A"
|
||
|
||
|
||
@patch("duty_teller.api.app.build_personal_ics")
|
||
@patch("duty_teller.api.app.get_duties_for_user")
|
||
@patch("duty_teller.api.app.get_user_by_calendar_token")
|
||
def test_calendar_ical_ignores_unknown_query_params(
|
||
mock_get_user, mock_get_duties, mock_build_ics, client
|
||
):
|
||
"""Unknown query params (e.g. events=all) are ignored; response is duty-only."""
|
||
from types import SimpleNamespace
|
||
|
||
from duty_teller.cache import ics_calendar_cache
|
||
|
||
ics_calendar_cache.invalidate(("personal_ics", 1))
|
||
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",
|
||
)
|
||
# get_duties_for_user returns (Duty, full_name, phone, username) tuples.
|
||
mock_get_duties.return_value = [(duty, "User A", None, None)]
|
||
mock_build_ics.return_value = b"BEGIN:VCALENDAR\r\nEND:VCALENDAR"
|
||
token = "z" * 43
|
||
|
||
r = client.get(f"/api/calendar/ical/{token}.ics", params={"events": "all"})
|
||
assert r.status_code == 200
|
||
mock_get_duties.assert_called_once_with(
|
||
ANY, 1, from_date=ANY, to_date=ANY, event_types=["duty"]
|
||
)
|
||
|
||
|
||
# --- /api/calendar-events ---
|
||
|
||
|
||
@patch("duty_teller.api.dependencies.config.MINI_APP_SKIP_AUTH", False)
|
||
def test_calendar_events_403_without_init_data(client):
|
||
"""Without X-Telegram-Init-Data and without MINI_APP_SKIP_AUTH → 403."""
|
||
r = client.get(
|
||
"/api/calendar-events",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
)
|
||
assert r.status_code == 403
|
||
|
||
|
||
@patch("duty_teller.api.app.config.EXTERNAL_CALENDAR_ICS_URL", "")
|
||
@patch("duty_teller.api.app.config.MINI_APP_SKIP_AUTH", True)
|
||
def test_calendar_events_empty_url_returns_empty_list(client):
|
||
"""When EXTERNAL_CALENDAR_ICS_URL is empty, GET /api/calendar-events returns [] without fetch."""
|
||
with patch("duty_teller.api.app.get_calendar_events") as mock_get:
|
||
r = client.get(
|
||
"/api/calendar-events",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json() == []
|
||
mock_get.assert_not_called()
|
||
|
||
|
||
@patch(
|
||
"duty_teller.api.app.config.EXTERNAL_CALENDAR_ICS_URL",
|
||
"https://example.com/cal.ics",
|
||
)
|
||
@patch("duty_teller.api.app.config.MINI_APP_SKIP_AUTH", True)
|
||
def test_calendar_events_200_returns_list_with_date_summary(client):
|
||
"""GET /api/calendar-events with auth and URL set returns list of {date, summary}."""
|
||
with patch("duty_teller.api.app.get_calendar_events") as mock_get:
|
||
mock_get.return_value = [
|
||
{"date": "2025-01-15", "summary": "Team meeting"},
|
||
{"date": "2025-01-20", "summary": "Review"},
|
||
]
|
||
r = client.get(
|
||
"/api/calendar-events",
|
||
params={"from": "2025-01-01", "to": "2025-01-31"},
|
||
)
|
||
assert r.status_code == 200
|
||
data = r.json()
|
||
assert len(data) == 2
|
||
assert data[0]["date"] == "2025-01-15"
|
||
assert data[0]["summary"] == "Team meeting"
|
||
assert data[1]["date"] == "2025-01-20"
|
||
assert data[1]["summary"] == "Review"
|
||
mock_get.assert_called_once_with(
|
||
"https://example.com/cal.ics",
|
||
from_date="2025-01-01",
|
||
to_date="2025-01-31",
|
||
)
|