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.
231 lines
7.4 KiB
Python
231 lines
7.4 KiB
Python
"""Integration tests for duty-schedule import (parser + repo, no bot)."""
|
||
|
||
from datetime import date
|
||
|
||
import pytest
|
||
|
||
from duty_teller.db import init_db
|
||
from duty_teller.db.repository import get_duties
|
||
from duty_teller.db.session import get_session, session_scope
|
||
from duty_teller.importers.duty_schedule import (
|
||
DutyScheduleEntry,
|
||
DutyScheduleResult,
|
||
parse_duty_schedule,
|
||
)
|
||
from duty_teller.services.import_service import run_import
|
||
|
||
|
||
@pytest.fixture
|
||
def db_url():
|
||
return "sqlite:///:memory:"
|
||
|
||
|
||
def _dispose_global_engine(session_module):
|
||
"""Dispose global engine so connections are closed, then clear cache."""
|
||
if session_module._engine is not None:
|
||
session_module._engine.dispose()
|
||
session_module._engine = None
|
||
session_module._SessionLocal = None
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_db_session(db_url):
|
||
"""Ensure each test uses a fresh engine for :memory: (clear global cache for test URL)."""
|
||
import duty_teller.db.session as session_module
|
||
|
||
_dispose_global_engine(session_module)
|
||
init_db(db_url)
|
||
yield
|
||
_dispose_global_engine(session_module)
|
||
|
||
|
||
def test_import_creates_users_and_duties(db_url):
|
||
"""Import creates users by full_name and correct duty records."""
|
||
result = DutyScheduleResult(
|
||
start_date=date(2026, 2, 16),
|
||
end_date=date(2026, 2, 18),
|
||
entries=[
|
||
DutyScheduleEntry(
|
||
full_name="Ivanov I.I.",
|
||
duty_dates=[date(2026, 2, 16), date(2026, 2, 18)],
|
||
unavailable_dates=[],
|
||
vacation_dates=[],
|
||
),
|
||
DutyScheduleEntry(
|
||
full_name="Petrov P.P.",
|
||
duty_dates=[date(2026, 2, 17)],
|
||
unavailable_dates=[],
|
||
vacation_dates=[],
|
||
),
|
||
],
|
||
)
|
||
with session_scope(db_url) as session:
|
||
num_users, num_duty, num_unav, num_vac = run_import(session, result, 6, 0)
|
||
assert num_users == 2
|
||
assert num_duty == 3
|
||
assert num_unav == 0
|
||
assert num_vac == 0
|
||
|
||
session = get_session(db_url)
|
||
try:
|
||
duties = get_duties(session, "2026-02-16", "2026-02-19")
|
||
finally:
|
||
session.close()
|
||
|
||
assert len(duties) == 3
|
||
starts = {d[0].start_at for d in duties}
|
||
assert "2026-02-16T06:00:00Z" in starts
|
||
assert "2026-02-17T06:00:00Z" in starts
|
||
assert "2026-02-18T06:00:00Z" in starts
|
||
for d, *_ in duties:
|
||
assert d.event_type == "duty"
|
||
|
||
|
||
def test_import_replaces_duties_in_range(db_url):
|
||
"""Re-importing same range replaces old duties."""
|
||
result1 = DutyScheduleResult(
|
||
start_date=date(2026, 2, 16),
|
||
end_date=date(2026, 2, 17),
|
||
entries=[
|
||
DutyScheduleEntry(
|
||
full_name="Sidorov",
|
||
duty_dates=[date(2026, 2, 16), date(2026, 2, 17)],
|
||
unavailable_dates=[],
|
||
vacation_dates=[],
|
||
)
|
||
],
|
||
)
|
||
with session_scope(db_url) as session:
|
||
run_import(session, result1, 9, 0)
|
||
|
||
session = get_session(db_url)
|
||
try:
|
||
duties_first = get_duties(session, "2026-02-16", "2026-02-18")
|
||
finally:
|
||
session.close()
|
||
assert len(duties_first) == 2
|
||
|
||
result2 = DutyScheduleResult(
|
||
start_date=date(2026, 2, 16),
|
||
end_date=date(2026, 2, 17),
|
||
entries=[
|
||
DutyScheduleEntry(
|
||
full_name="Sidorov",
|
||
duty_dates=[date(2026, 2, 17)],
|
||
unavailable_dates=[],
|
||
vacation_dates=[],
|
||
)
|
||
],
|
||
)
|
||
with session_scope(db_url) as session:
|
||
run_import(session, result2, 9, 0)
|
||
|
||
session = get_session(db_url)
|
||
try:
|
||
duties_second = get_duties(session, "2026-02-16", "2026-02-18")
|
||
finally:
|
||
session.close()
|
||
assert len(duties_second) == 1
|
||
assert duties_second[0][0].start_at == "2026-02-17T09:00:00Z"
|
||
|
||
|
||
def test_import_full_flow_parse_then_import(db_url):
|
||
"""Parse real-looking JSON then run import."""
|
||
raw = (
|
||
'{"meta": {"start_date": "2026-02-16"}, '
|
||
'"schedule": [{"name": "Alexey A.", "duty": "\u0431; ; \u0432"}]}'
|
||
).encode("utf-8")
|
||
parsed = parse_duty_schedule(raw)
|
||
with session_scope(db_url) as session:
|
||
num_users, num_duty, num_unav, num_vac = run_import(session, parsed, 6, 0)
|
||
assert num_users == 1
|
||
assert num_duty == 2
|
||
assert num_unav == 0
|
||
assert num_vac == 0
|
||
|
||
session = get_session(db_url)
|
||
try:
|
||
duties = get_duties(session, "2026-02-16", "2026-02-19")
|
||
finally:
|
||
session.close()
|
||
assert len(duties) == 2
|
||
assert duties[0][1] == "Alexey A."
|
||
assert duties[0][0].event_type == "duty"
|
||
|
||
|
||
def test_import_event_types_unavailable_vacation(db_url):
|
||
"""Import creates duty, unavailable (full day), vacation (periods). Unavailable: same-day 00:00–23:59. Three consecutive vacation days → one record."""
|
||
result = DutyScheduleResult(
|
||
start_date=date(2026, 2, 16),
|
||
end_date=date(2026, 2, 20),
|
||
entries=[
|
||
DutyScheduleEntry(
|
||
full_name="Mixed User",
|
||
duty_dates=[date(2026, 2, 16)],
|
||
unavailable_dates=[date(2026, 2, 17)],
|
||
vacation_dates=[
|
||
date(2026, 2, 18),
|
||
date(2026, 2, 19),
|
||
date(2026, 2, 20),
|
||
],
|
||
),
|
||
],
|
||
)
|
||
with session_scope(db_url) as session:
|
||
num_users, num_duty, num_unav, num_vac = run_import(session, result, 6, 0)
|
||
assert num_users == 1
|
||
assert num_duty == 1 and num_unav == 1 and num_vac == 1
|
||
|
||
session = get_session(db_url)
|
||
try:
|
||
duties = get_duties(session, "2026-02-16", "2026-02-21")
|
||
finally:
|
||
session.close()
|
||
assert len(duties) == 3
|
||
types = {d[0].event_type for d in duties}
|
||
assert types == {"duty", "unavailable", "vacation"}
|
||
|
||
by_type = {d[0].event_type: d[0] for d in duties}
|
||
unav = by_type["unavailable"]
|
||
assert unav.start_at == "2026-02-17T00:00:00Z"
|
||
assert unav.end_at == "2026-02-17T23:59:59Z"
|
||
vac = by_type["vacation"]
|
||
assert vac.start_at == "2026-02-18T00:00:00Z"
|
||
assert vac.end_at == "2026-02-20T23:59:59Z"
|
||
|
||
|
||
def test_import_vacation_with_gap_two_periods(db_url):
|
||
"""Vacation dates with a gap (17, 18, 20 Feb) → two records: 17–18 and 20."""
|
||
result = DutyScheduleResult(
|
||
start_date=date(2026, 2, 16),
|
||
end_date=date(2026, 2, 21),
|
||
entries=[
|
||
DutyScheduleEntry(
|
||
full_name="Vacation User",
|
||
duty_dates=[],
|
||
unavailable_dates=[],
|
||
vacation_dates=[
|
||
date(2026, 2, 17),
|
||
date(2026, 2, 18),
|
||
date(2026, 2, 20),
|
||
],
|
||
),
|
||
],
|
||
)
|
||
with session_scope(db_url) as session:
|
||
num_users, num_duty, num_unav, num_vac = run_import(session, result, 6, 0)
|
||
assert num_users == 1
|
||
assert num_duty == 0 and num_unav == 0 and num_vac == 2
|
||
|
||
session = get_session(db_url)
|
||
try:
|
||
duties = get_duties(session, "2026-02-16", "2026-02-21")
|
||
finally:
|
||
session.close()
|
||
vacation_records = [d[0] for d in duties if d[0].event_type == "vacation"]
|
||
assert len(vacation_records) == 2
|
||
starts = sorted(r.start_at for r in vacation_records)
|
||
ends = sorted(r.end_at for r in vacation_records)
|
||
assert starts == ["2026-02-17T00:00:00Z", "2026-02-20T00:00:00Z"]
|
||
assert ends == ["2026-02-18T23:59:59Z", "2026-02-20T23:59:59Z"]
|