Nikolay Tatarinov
aa89494bd5
All checks were successful
CI / lint-and-test (push) Successful in 22s
- Added support for filtering calendar events by type in the ICS generation API endpoint, allowing users to specify whether to include only duty shifts or all event types (duty, unavailable, vacation). - Updated the `get_duties_for_user` function to accept an optional `event_types` parameter, enabling more flexible data retrieval based on user preferences. - Enhanced unit tests to cover the new event type filtering functionality, ensuring correct behavior and reliability of the ICS generation process.
197 lines
6.3 KiB
Python
197 lines
6.3 KiB
Python
"""Tests for duty_teller.services.group_duty_pin_service."""
|
|
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from duty_teller.db.models import Base, Duty, User
|
|
from duty_teller.services import group_duty_pin_service as svc
|
|
|
|
|
|
@pytest.fixture
|
|
def session():
|
|
"""In-memory SQLite session with all models."""
|
|
engine = create_engine(
|
|
"sqlite:///:memory:", connect_args={"check_same_thread": False}
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
Session = sessionmaker(bind=engine, autocommit=False, autoflush=False)
|
|
s = Session()
|
|
try:
|
|
yield s
|
|
finally:
|
|
s.close()
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def user(session):
|
|
"""Create a user in DB."""
|
|
u = User(
|
|
telegram_user_id=123,
|
|
full_name="Test User",
|
|
username="testuser",
|
|
first_name="Test",
|
|
last_name="User",
|
|
phone="+79001234567",
|
|
)
|
|
session.add(u)
|
|
session.commit()
|
|
session.refresh(u)
|
|
return u
|
|
|
|
|
|
@pytest.fixture
|
|
def duty(session, user):
|
|
"""Create a duty in DB (event_type='duty')."""
|
|
d = Duty(
|
|
user_id=user.id,
|
|
start_at="2025-02-20T06:00:00Z",
|
|
end_at="2025-02-21T06:00:00Z",
|
|
event_type="duty",
|
|
)
|
|
session.add(d)
|
|
session.commit()
|
|
session.refresh(d)
|
|
return d
|
|
|
|
|
|
class TestFormatDutyMessage:
|
|
"""Tests for format_duty_message."""
|
|
|
|
def test_none_duty_returns_no_duty(self):
|
|
user = SimpleNamespace(full_name="U")
|
|
with patch("duty_teller.services.group_duty_pin_service.t") as mock_t:
|
|
mock_t.return_value = "No duty"
|
|
result = svc.format_duty_message(None, user, "Europe/Moscow", "en")
|
|
assert result == "No duty"
|
|
mock_t.assert_called_with("en", "duty.no_duty")
|
|
|
|
def test_none_user_returns_no_duty(self):
|
|
duty = SimpleNamespace(
|
|
start_at="2025-01-15T09:00:00Z", end_at="2025-01-15T18:00:00Z"
|
|
)
|
|
with patch("duty_teller.services.group_duty_pin_service.t") as mock_t:
|
|
mock_t.return_value = "No duty"
|
|
result = svc.format_duty_message(duty, None, "Europe/Moscow", "en")
|
|
assert result == "No duty"
|
|
|
|
def test_with_duty_and_user_returns_formatted(self):
|
|
duty = SimpleNamespace(
|
|
start_at="2025-01-15T09:00:00Z",
|
|
end_at="2025-01-15T18:00:00Z",
|
|
)
|
|
user = SimpleNamespace(
|
|
full_name="Иван Иванов",
|
|
phone="+79001234567",
|
|
username="ivan",
|
|
)
|
|
with patch("duty_teller.services.group_duty_pin_service.t") as mock_t:
|
|
mock_t.side_effect = lambda lang, key: "Duty" if key == "duty.label" else ""
|
|
result = svc.format_duty_message(duty, user, "Europe/Moscow", "ru")
|
|
assert "Иван Иванов" in result
|
|
assert "+79001234567" in result or "79001234567" in result
|
|
assert "@ivan" in result
|
|
assert "Duty" in result
|
|
|
|
|
|
class TestGetDutyMessageText:
|
|
"""Tests for get_duty_message_text."""
|
|
|
|
def test_no_current_duty_returns_no_duty(self, session):
|
|
with patch(
|
|
"duty_teller.services.group_duty_pin_service.get_current_duty",
|
|
return_value=None,
|
|
):
|
|
with patch("duty_teller.services.group_duty_pin_service.t") as mock_t:
|
|
mock_t.return_value = "No duty"
|
|
result = svc.get_duty_message_text(session, "Europe/Moscow", "en")
|
|
assert result == "No duty"
|
|
|
|
def test_with_current_duty_returns_formatted(self, session, duty, user):
|
|
with patch(
|
|
"duty_teller.services.group_duty_pin_service.get_current_duty",
|
|
return_value=(duty, user),
|
|
):
|
|
with patch("duty_teller.services.group_duty_pin_service.t") as mock_t:
|
|
mock_t.side_effect = lambda lang, key: (
|
|
"Duty" if key == "duty.label" else "No duty"
|
|
)
|
|
result = svc.get_duty_message_text(session, "Europe/Moscow", "en")
|
|
assert "Test User" in result
|
|
assert "Duty" in result
|
|
|
|
|
|
class TestGetNextShiftEndUtc:
|
|
"""Tests for get_next_shift_end_utc."""
|
|
|
|
def test_no_next_shift_returns_none(self, session):
|
|
with patch(
|
|
"duty_teller.services.group_duty_pin_service.get_next_shift_end",
|
|
return_value=None,
|
|
):
|
|
result = svc.get_next_shift_end_utc(session)
|
|
assert result is None
|
|
|
|
def test_has_next_shift_returns_naive_utc(self, session):
|
|
naive = datetime(2025, 2, 21, 6, 0, 0)
|
|
with patch(
|
|
"duty_teller.services.group_duty_pin_service.get_next_shift_end",
|
|
return_value=naive,
|
|
):
|
|
result = svc.get_next_shift_end_utc(session)
|
|
assert result == naive
|
|
|
|
|
|
class TestSavePin:
|
|
"""Tests for save_pin."""
|
|
|
|
def test_save_pin_creates_record(self, session):
|
|
svc.save_pin(session, chat_id=100, message_id=42)
|
|
mid = svc.get_message_id(session, 100)
|
|
assert mid == 42
|
|
|
|
def test_save_pin_updates_existing(self, session):
|
|
svc.save_pin(session, chat_id=100, message_id=42)
|
|
svc.save_pin(session, chat_id=100, message_id=99)
|
|
mid = svc.get_message_id(session, 100)
|
|
assert mid == 99
|
|
|
|
|
|
class TestDeletePin:
|
|
"""Tests for delete_pin."""
|
|
|
|
def test_delete_pin_removes_record(self, session):
|
|
svc.save_pin(session, chat_id=200, message_id=1)
|
|
assert svc.get_message_id(session, 200) == 1
|
|
svc.delete_pin(session, chat_id=200)
|
|
assert svc.get_message_id(session, 200) is None
|
|
|
|
|
|
class TestGetMessageId:
|
|
"""Tests for get_message_id."""
|
|
|
|
def test_no_pin_returns_none(self, session):
|
|
assert svc.get_message_id(session, 999) is None
|
|
|
|
def test_after_save_returns_message_id(self, session):
|
|
svc.save_pin(session, chat_id=300, message_id=77)
|
|
assert svc.get_message_id(session, 300) == 77
|
|
|
|
|
|
class TestGetAllPinChatIds:
|
|
"""Tests for get_all_pin_chat_ids."""
|
|
|
|
def test_empty_returns_empty_list(self, session):
|
|
assert svc.get_all_pin_chat_ids(session) == []
|
|
|
|
def test_returns_all_chat_ids_with_pins(self, session):
|
|
svc.save_pin(session, chat_id=10, message_id=1)
|
|
svc.save_pin(session, chat_id=20, message_id=2)
|
|
chat_ids = svc.get_all_pin_chat_ids(session)
|
|
assert set(chat_ids) == {10, 20}
|