from __future__ import annotations from datetime import datetime, time import pytest from streamd.localize.localized_shard import LocalizedShard from streamd.timesheet.configuration import ( TIMESHEET_DIMENSION_NAME, TimesheetPointType, ) from streamd.timesheet.extract import extract_timesheets from streamd.timesheet.timecard import SpecialDayType, Timecard, Timesheet def point(at: datetime, type: TimesheetPointType) -> LocalizedShard: """ Create a minimal LocalizedShard that will be interpreted as a timesheet point. Note: The extract pipeline uses set-dimension filtering; we therefore ensure the timesheet dimension is set in `location`. """ return LocalizedShard( moment=at, markers=["Timesheet"], tags=[], start_line=1, end_line=1, children=[], location={TIMESHEET_DIMENSION_NAME: type.value, "file": "dummy.md"}, ) class TestExtractTimesheets: def test_single_work_block(self): day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=9, minute=0), TimesheetPointType.Card), point(day.replace(hour=17, minute=30), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=False, special_day_type=None, timecards=[Timecard(from_time=time(9, 0), to_time=time(17, 30))], ) ] def test_three_work_blocks_separated_by_breaks(self): day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=7, minute=15), TimesheetPointType.Card), point(day.replace(hour=12, minute=0), TimesheetPointType.Break), point(day.replace(hour=12, minute=45), TimesheetPointType.Card), point(day.replace(hour=15, minute=0), TimesheetPointType.Break), point(day.replace(hour=16, minute=0), TimesheetPointType.Card), point(day.replace(hour=17, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=False, special_day_type=None, timecards=[ Timecard(from_time=time(7, 15), to_time=time(12, 0)), Timecard(from_time=time(12, 45), to_time=time(15, 0)), Timecard(from_time=time(16, 0), to_time=time(17, 0)), ], ) ] def test_input_order_is_not_required_within_a_day(self): """ Points may come unsorted; extraction should sort by timestamp within a day. """ day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=15, minute=0), TimesheetPointType.Break), point(day.replace(hour=7, minute=15), TimesheetPointType.Card), point(day.replace(hour=12, minute=0), TimesheetPointType.Break), point(day.replace(hour=12, minute=45), TimesheetPointType.Card), point(day.replace(hour=17, minute=0), TimesheetPointType.Break), point(day.replace(hour=16, minute=0), TimesheetPointType.Card), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=False, special_day_type=None, timecards=[ Timecard(from_time=time(7, 15), to_time=time(12, 0)), Timecard(from_time=time(12, 45), to_time=time(15, 0)), Timecard(from_time=time(16, 0), to_time=time(17, 0)), ], ) ] def test_groups_by_day(self): """ If points span multiple days, we should get one Timesheet per day. """ day1 = datetime(2026, 2, 1, 0, 0, 0) day2 = datetime(2026, 2, 2, 0, 0, 0) shards = [ point(day2.replace(hour=10, minute=0), TimesheetPointType.Card), point(day2.replace(hour=18, minute=0), TimesheetPointType.Break), point(day1.replace(hour=9, minute=0), TimesheetPointType.Card), point(day1.replace(hour=17, minute=0), TimesheetPointType.Break), ] # Note: current implementation groups by date using `itertools.groupby` on the # incoming order; to be robust, we pass day1 points first, then day2 points. # This asserts the intended behavior. shards = [ point(day1.replace(hour=9, minute=0), TimesheetPointType.Card), point(day1.replace(hour=17, minute=0), TimesheetPointType.Break), point(day2.replace(hour=10, minute=0), TimesheetPointType.Card), point(day2.replace(hour=18, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day1.date(), is_sick_leave=False, special_day_type=None, timecards=[Timecard(from_time=time(9, 0), to_time=time(17, 0))], ), Timesheet( date=day2.date(), is_sick_leave=False, special_day_type=None, timecards=[Timecard(from_time=time(10, 0), to_time=time(18, 0))], ), ] def test_day_with_only_special_day_type_vacation(self): """ A day can be marked as Vacation without timecards; it should still produce a Timesheet. """ day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=8, minute=0), TimesheetPointType.Vacation), point(day.replace(hour=9, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=False, special_day_type=SpecialDayType.Vacation, timecards=[], ) ] def test_day_with_only_special_day_type_holiday(self): day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=8, minute=0), TimesheetPointType.Holiday), point(day.replace(hour=9, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=False, special_day_type=SpecialDayType.Holiday, timecards=[], ) ] def test_day_with_only_special_day_type_undertime(self): day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=8, minute=0), TimesheetPointType.Undertime), point(day.replace(hour=9, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=False, special_day_type=SpecialDayType.Undertime, timecards=[], ) ] def test_day_with_sick_leave_and_timecards(self): """ SickLeave should set the flag but not prevent timecard aggregation. """ day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=7, minute=30), TimesheetPointType.SickLeave), point(day.replace(hour=9, minute=0), TimesheetPointType.Card), point(day.replace(hour=12, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=True, special_day_type=None, timecards=[Timecard(from_time=time(9, 0), to_time=time(12, 0))], ) ] def test_day_with_sick_leave_only(self): """ A day with only SickLeave should still produce a Timesheet (no timecards). """ day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=8, minute=0), TimesheetPointType.SickLeave), point(day.replace(hour=9, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == [ Timesheet( date=day.date(), is_sick_leave=True, special_day_type=None, timecards=[], ) ] def test_empty_input(self): assert extract_timesheets([]) == [] def test_day_with_only_cards_and_no_break_is_invalid(self): """ A day ending 'in work' (last point not a Break) should raise. """ day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=9, minute=0), TimesheetPointType.Card), point(day.replace(hour=12, minute=0), TimesheetPointType.Card), ] with pytest.raises(ValueError, match=r"Last Timecard of .* is not a break"): _ = extract_timesheets(shards) def test_two_special_day_types_same_day_is_invalid(self): """ A day cannot be both Vacation and Holiday (or any two distinct special types). """ day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=8, minute=0), TimesheetPointType.Vacation), point(day.replace(hour=8, minute=5), TimesheetPointType.Holiday), point(day.replace(hour=9, minute=0), TimesheetPointType.Break), ] with pytest.raises(ValueError, match=r"is both .* and .*"): _ = extract_timesheets(shards) def test_points_with_mixed_dates_inside_one_group_raises(self): """ Defensive: if aggregation receives points spanning multiple dates for a single day, it should raise. (This can occur if higher-level grouping is incorrect.) """ day1 = datetime(2026, 2, 1, 0, 0, 0) day2 = datetime(2026, 2, 2, 0, 0, 0) shards = [ point(day1.replace(hour=9, minute=0), TimesheetPointType.Card), point(day2.replace(hour=9, minute=30), TimesheetPointType.Break), ] with pytest.raises(ValueError, match=r"Last Timecard of .* is not a break"): _ = extract_timesheets(shards) def test_day_with_only_breaks_is_ignored(self): """ A day with no timecards and no sick/special markers should not emit a Timesheet. """ day = datetime(2026, 2, 1, 0, 0, 0) shards = [ point(day.replace(hour=12, minute=0), TimesheetPointType.Break), point(day.replace(hour=13, minute=0), TimesheetPointType.Break), ] assert extract_timesheets(shards) == []