kfickel/streamd
1
0
Fork 0
Code Issues 10 Pull requests 3 Projects 1 Wiki Activity Actions

refactor: rename test/ to tests/

Browse source
This commit is contained in:
Konstantin Fickel 2026-02-15 17:30:44 +01:00
parent af2debc19b
commit f9ed0463f7
Signed by: kfickel
GPG key ID: A793722F9933C1A5
6 changed files with 16 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

288
test/timesheet/test_extract_timesheets.py
View file

@ -1,288 +0,0 @@
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) == []