Nikolay Tatarinov
e25eb7be2f
Some checks failed
CI / lint-and-test (push) Failing after 11s
- Upgraded `pytest-asyncio` to version 1.0 to ensure compatibility with the latest features and improvements. - Increased the coverage threshold in pytest configuration to 80%, enhancing the quality assurance process. - Added a new `conftest.py` file to manage shared fixtures and improve test organization. - Introduced multiple new test files to cover various components, ensuring comprehensive test coverage across the application. - Updated the `.coverage` file to reflect the latest coverage metrics.
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, timezone
|
|
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, GroupDutyPin, 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}
|