Nikolay Tatarinov
e3240d0981
- Added `bot_username` to settings for dynamic retrieval of the bot's username. - Implemented `_resolve_bot_username` function to fetch the bot's username if not set, improving user experience in group chats. - Updated `DutyWithUser` schema to include optional `phone` and `username` fields for enhanced duty information. - Enhanced API responses to include contact details for users, ensuring better communication. - Introduced a new current duty view in the web app, displaying active duty information along with contact options. - Updated CSS styles for better presentation of contact information in duty cards. - Added unit tests to verify the inclusion of contact details in API responses and the functionality of the current duty view.
203 lines
6.8 KiB
Python
203 lines
6.8 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):
|
|
"""Formatted message includes time range and full name only; no contact info (phone/username)."""
|
|
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 "Duty" in result
|
|
# Contact info is restricted to Mini App; not shown in pinned group message.
|
|
assert "+79001234567" not in result and "79001234567" not in result
|
|
assert "@ivan" not in result
|
|
|
|
|
|
class TestGetDutyMessageText:
|
|
"""Tests for get_duty_message_text."""
|
|
|
|
def test_no_current_duty_returns_no_duty(self, session):
|
|
svc.duty_pin_cache.invalidate_pattern(("duty_message_text",))
|
|
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):
|
|
svc.duty_pin_cache.invalidate_pattern(("duty_message_text",))
|
|
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):
|
|
svc.duty_pin_cache.invalidate(("next_shift_end",))
|
|
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):
|
|
svc.duty_pin_cache.invalidate(("next_shift_end",))
|
|
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}
|