Nikolay Tatarinov
c390a4dd6e
- Added new API endpoints for admin features: `GET /api/admin/me`, `GET /api/admin/users`, and `PATCH /api/admin/duties/:id` to manage user duties. - Introduced `UserForAdmin` and `AdminDutyReassignBody` schemas for handling admin-related data. - Updated documentation to include Mini App design guidelines and admin panel functionalities. - Enhanced tests for admin API to ensure proper access control and functionality. - Improved error handling and localization for admin actions.
297 lines
9.2 KiB
Python
297 lines
9.2 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_duty_by_id,
|
|
get_or_create_user,
|
|
get_or_create_user_by_full_name,
|
|
get_users_for_admin,
|
|
insert_duty,
|
|
update_duty_user,
|
|
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_get_duty_by_id_returns_duty(session, user_a):
|
|
"""get_duty_by_id returns the duty when it exists."""
|
|
duty = insert_duty(
|
|
session, user_a.id, "2026-02-01T09:00:00Z", "2026-02-01T18:00:00Z"
|
|
)
|
|
found = get_duty_by_id(session, duty.id)
|
|
assert found is not None
|
|
assert found.id == duty.id
|
|
assert found.user_id == user_a.id
|
|
assert found.start_at == "2026-02-01T09:00:00Z"
|
|
|
|
|
|
def test_get_duty_by_id_returns_none_when_missing(session):
|
|
"""get_duty_by_id returns None for non-existent id."""
|
|
assert get_duty_by_id(session, 99999) is None
|
|
|
|
|
|
def test_update_duty_user_changes_user(session, user_a):
|
|
"""update_duty_user updates user_id and returns the duty."""
|
|
user_b = get_or_create_user_by_full_name(session, "User B")
|
|
duty = insert_duty(
|
|
session, user_a.id, "2026-02-01T09:00:00Z", "2026-02-01T18:00:00Z"
|
|
)
|
|
updated = update_duty_user(session, duty.id, user_b.id, commit=True)
|
|
assert updated is not None
|
|
assert updated.id == duty.id
|
|
assert updated.user_id == user_b.id
|
|
session.refresh(duty)
|
|
assert duty.user_id == user_b.id
|
|
|
|
|
|
def test_update_duty_user_returns_none_when_duty_missing(session, user_a):
|
|
"""update_duty_user returns None when duty does not exist."""
|
|
assert update_duty_user(session, 99999, user_a.id, commit=True) is None
|
|
|
|
|
|
def test_get_users_for_admin_returns_all_ordered_by_full_name(session, user_a):
|
|
"""get_users_for_admin returns all users ordered by full_name."""
|
|
get_or_create_user_by_full_name(session, "Alice")
|
|
get_or_create_user_by_full_name(session, "Борис")
|
|
users = get_users_for_admin(session)
|
|
assert len(users) >= 3
|
|
full_names = [u.full_name for u in users]
|
|
assert full_names == sorted(full_names)
|
|
|
|
|
|
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"
|