"""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"]