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.
248 lines
7.4 KiB
Python
248 lines
7.4 KiB
Python
"""Tests for delete_duties_in_range, get_or_create_user_by_full_name, name_manually_edited."""
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from duty_teller.db.models import Base, User
|
|
from duty_teller.db.repository import (
|
|
delete_duties_in_range,
|
|
get_duties,
|
|
get_duties_for_user,
|
|
get_or_create_user,
|
|
get_or_create_user_by_full_name,
|
|
insert_duty,
|
|
update_user_display_name,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def session():
|
|
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_a(session):
|
|
u = User(
|
|
telegram_user_id=None,
|
|
full_name="User A",
|
|
username=None,
|
|
first_name=None,
|
|
last_name=None,
|
|
)
|
|
session.add(u)
|
|
session.commit()
|
|
session.refresh(u)
|
|
return u
|
|
|
|
|
|
def test_get_or_create_user_by_full_name_creates(session):
|
|
u = get_or_create_user_by_full_name(session, "Новый Пользователь")
|
|
assert u.id is not None
|
|
assert u.full_name == "Новый Пользователь"
|
|
assert u.telegram_user_id is None
|
|
assert u.name_manually_edited is True
|
|
|
|
|
|
def test_get_or_create_user_by_full_name_returns_existing(session, user_a):
|
|
u = get_or_create_user_by_full_name(session, "User A")
|
|
assert u.id == user_a.id
|
|
assert u.full_name == "User A"
|
|
|
|
|
|
def test_delete_duties_in_range_removes_only_in_range(session, user_a):
|
|
# Duties: 2026-02-01 06:00 - 2026-02-02 06:00; 2026-02-15 - 2026-02-16; 2026-02-28 - 2026-03-01
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-02-01T06:00:00Z",
|
|
"2026-02-02T06:00:00Z",
|
|
)
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-02-15T06:00:00Z",
|
|
"2026-02-16T06:00:00Z",
|
|
)
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-02-28T06:00:00Z",
|
|
"2026-03-01T06:00:00Z",
|
|
)
|
|
deleted = delete_duties_in_range(session, user_a.id, "2026-02-10", "2026-02-20")
|
|
assert deleted == 1
|
|
remaining = get_duties(session, "2026-01-01", "2026-03-31")
|
|
assert len(remaining) == 2
|
|
starts = [d[0].start_at for d in remaining]
|
|
assert "2026-02-01T06:00:00Z" in starts
|
|
assert "2026-02-28T06:00:00Z" in starts
|
|
assert "2026-02-15T06:00:00Z" not in starts
|
|
|
|
|
|
def test_delete_duties_in_range_other_user_unchanged(session, user_a):
|
|
user_b = get_or_create_user_by_full_name(session, "User B")
|
|
insert_duty(session, user_a.id, "2026-02-10T06:00:00Z", "2026-02-11T06:00:00Z")
|
|
insert_duty(session, user_b.id, "2026-02-10T06:00:00Z", "2026-02-11T06:00:00Z")
|
|
delete_duties_in_range(session, user_a.id, "2026-02-01", "2026-02-28")
|
|
remaining = get_duties(session, "2026-02-01", "2026-02-28")
|
|
assert len(remaining) == 1
|
|
assert remaining[0][1] == "User B"
|
|
|
|
|
|
def test_get_duties_includes_duty_starting_on_last_day_of_range(session, user_a):
|
|
"""Duty starting on to_date (e.g. 2026-01-31T09:00:00Z) must be included when to_date is 2026-01-31."""
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-01-31T09:00:00Z",
|
|
"2026-02-01T09:00:00Z",
|
|
)
|
|
rows = get_duties(session, "2026-01-01", "2026-01-31")
|
|
assert len(rows) == 1
|
|
assert rows[0][0].start_at == "2026-01-31T09:00:00Z"
|
|
assert rows[0][1] == "User A"
|
|
|
|
|
|
def test_get_duties_for_user_event_types_duty_returns_only_duty(session, user_a):
|
|
"""get_duties_for_user(..., event_types=["duty"]) returns only duty records."""
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-02-01T09:00:00Z",
|
|
"2026-02-01T18:00:00Z",
|
|
event_type="duty",
|
|
)
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-02-02T09:00:00Z",
|
|
"2026-02-02T18:00:00Z",
|
|
event_type="unavailable",
|
|
)
|
|
rows = get_duties_for_user(
|
|
session, user_a.id, "2026-02-01", "2026-02-28", event_types=["duty"]
|
|
)
|
|
assert len(rows) == 1
|
|
assert rows[0][0].event_type == "duty"
|
|
assert rows[0][1] == "User A"
|
|
|
|
|
|
def test_get_duties_for_user_event_types_none_returns_all(session, user_a):
|
|
"""get_duties_for_user(..., event_types=None) returns duty and unavailable."""
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-02-01T09:00:00Z",
|
|
"2026-02-01T18:00:00Z",
|
|
event_type="duty",
|
|
)
|
|
insert_duty(
|
|
session,
|
|
user_a.id,
|
|
"2026-02-02T09:00:00Z",
|
|
"2026-02-02T18:00:00Z",
|
|
event_type="unavailable",
|
|
)
|
|
rows = get_duties_for_user(
|
|
session, user_a.id, "2026-02-01", "2026-02-28", event_types=None
|
|
)
|
|
assert len(rows) == 2
|
|
types = {rows[0][0].event_type, rows[1][0].event_type}
|
|
assert types == {"duty", "unavailable"}
|
|
|
|
|
|
def test_get_or_create_user_overwrites_name_when_flag_false(session):
|
|
"""When name_manually_edited is False, second get_or_create_user overwrites name."""
|
|
u1 = get_or_create_user(
|
|
session,
|
|
telegram_user_id=100,
|
|
full_name="First Name",
|
|
username="user1",
|
|
first_name="First",
|
|
last_name="Name",
|
|
)
|
|
assert u1.full_name == "First Name"
|
|
assert u1.name_manually_edited is False
|
|
u2 = get_or_create_user(
|
|
session,
|
|
telegram_user_id=100,
|
|
full_name="Second Name",
|
|
username="user2",
|
|
first_name="Second",
|
|
last_name="Name",
|
|
)
|
|
assert u2.id == u1.id
|
|
assert u2.full_name == "Second Name"
|
|
assert u2.first_name == "Second"
|
|
assert u2.last_name == "Name"
|
|
assert u2.username == "user2"
|
|
|
|
|
|
def test_get_or_create_user_keeps_name_when_flag_true_updates_username(session):
|
|
"""When name_manually_edited is True, get_or_create_user keeps name but updates username."""
|
|
u1 = get_or_create_user(
|
|
session,
|
|
telegram_user_id=200,
|
|
full_name="Custom Name",
|
|
username="old_username",
|
|
first_name="Custom",
|
|
last_name="Name",
|
|
)
|
|
u1.name_manually_edited = True
|
|
session.commit()
|
|
session.refresh(u1)
|
|
u2 = get_or_create_user(
|
|
session,
|
|
telegram_user_id=200,
|
|
full_name="Telegram Name",
|
|
username="new_username",
|
|
first_name="Telegram",
|
|
last_name="Name",
|
|
)
|
|
assert u2.id == u1.id
|
|
assert u2.full_name == "Custom Name"
|
|
assert u2.first_name == "Custom"
|
|
assert u2.last_name == "Name"
|
|
assert u2.username == "new_username"
|
|
|
|
|
|
def test_update_user_display_name_sets_flag_then_get_or_create_user_keeps_name(session):
|
|
"""update_user_display_name sets name and flag; get_or_create_user then does not overwrite name."""
|
|
get_or_create_user(
|
|
session,
|
|
telegram_user_id=300,
|
|
full_name="Original",
|
|
username="u3",
|
|
first_name="Original",
|
|
last_name=None,
|
|
)
|
|
updated = update_user_display_name(
|
|
session, 300, "Manual Name", first_name="Manual", last_name="Name"
|
|
)
|
|
assert updated is not None
|
|
assert updated.full_name == "Manual Name"
|
|
assert updated.name_manually_edited is True
|
|
u_after_sync = get_or_create_user(
|
|
session,
|
|
telegram_user_id=300,
|
|
full_name="From Telegram",
|
|
username="new_u3",
|
|
first_name="From",
|
|
last_name="Telegram",
|
|
)
|
|
assert u_after_sync.full_name == "Manual Name"
|
|
assert u_after_sync.first_name == "Manual"
|
|
assert u_after_sync.last_name == "Name"
|
|
assert u_after_sync.username == "new_u3"
|