From 6a48739ebe633e5ddad7a372f506277d5bf18d57 Mon Sep 17 00:00:00 2001 From: Valentin Todorov Date: Mon, 22 Sep 2025 10:59:20 +0300 Subject: [PATCH] feat: add time calculation utilities --- src/cronpal/time_utils.py | 279 ++++++++++++++++++++++++++++++++++++++ tests/test_time_utils.py | 279 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 558 insertions(+) create mode 100644 src/cronpal/time_utils.py create mode 100644 tests/test_time_utils.py diff --git a/src/cronpal/time_utils.py b/src/cronpal/time_utils.py new file mode 100644 index 0000000..5f7bd98 --- /dev/null +++ b/src/cronpal/time_utils.py @@ -0,0 +1,279 @@ +"""Time calculation utilities for cron expressions.""" + +import calendar +from datetime import datetime, timedelta +from typing import Optional, Tuple + + +def get_next_minute(dt: datetime) -> datetime: + """ + Get the next minute from the given datetime. + + Args: + dt: The datetime to increment. + + Returns: + A datetime object representing the next minute. + """ + return dt + timedelta(minutes=1) + + +def get_next_hour(dt: datetime) -> datetime: + """ + Get the next hour from the given datetime (minute set to 0). + + Args: + dt: The datetime to increment. + + Returns: + A datetime object at the start of the next hour. + """ + next_hour = dt.replace(minute=0, second=0, microsecond=0) + return next_hour + timedelta(hours=1) + + +def get_next_day(dt: datetime) -> datetime: + """ + Get the next day from the given datetime (time set to 00:00). + + Args: + dt: The datetime to increment. + + Returns: + A datetime object at the start of the next day. + """ + next_day = dt.replace(hour=0, minute=0, second=0, microsecond=0) + return next_day + timedelta(days=1) + + +def get_next_month(dt: datetime) -> datetime: + """ + Get the first day of the next month. + + Args: + dt: The datetime to increment. + + Returns: + A datetime object at the start of the next month. + """ + # Calculate next month + year = dt.year + month = dt.month + 1 + + if month > 12: + month = 1 + year += 1 + + return datetime(year, month, 1, 0, 0, 0, 0) + + +def get_next_year(dt: datetime) -> datetime: + """ + Get the first day of the next year. + + Args: + dt: The datetime to increment. + + Returns: + A datetime object at the start of the next year. + """ + return datetime(dt.year + 1, 1, 1, 0, 0, 0, 0) + + +def get_days_in_month(year: int, month: int) -> int: + """ + Get the number of days in a given month. + + Args: + year: The year. + month: The month (1-12). + + Returns: + The number of days in the month. + """ + return calendar.monthrange(year, month)[1] + + +def is_leap_year(year: int) -> bool: + """ + Check if a year is a leap year. + + Args: + year: The year to check. + + Returns: + True if it's a leap year, False otherwise. + """ + return calendar.isleap(year) + + +def get_weekday(dt: datetime) -> int: + """ + Get the weekday for a datetime (0=Monday, 6=Sunday). + Converts to cron format (0=Sunday, 6=Saturday). + + Args: + dt: The datetime to check. + + Returns: + The weekday in cron format (0=Sunday, 6=Saturday). + """ + # Python's weekday: 0=Monday, 6=Sunday + # Cron's weekday: 0=Sunday, 6=Saturday + python_weekday = dt.weekday() + + if python_weekday == 6: # Sunday in Python + return 0 # Sunday in cron + else: + return python_weekday + 1 # Shift Monday-Saturday + + +def get_month_day_count(dt: datetime) -> int: + """ + Get the number of days in the current month of the datetime. + + Args: + dt: The datetime to check. + + Returns: + The number of days in the month. + """ + return get_days_in_month(dt.year, dt.month) + + +def normalize_datetime(dt: datetime) -> datetime: + """ + Normalize a datetime to remove seconds and microseconds. + + Args: + dt: The datetime to normalize. + + Returns: + A datetime with seconds and microseconds set to 0. + """ + return dt.replace(second=0, microsecond=0) + + +def round_to_next_minute(dt: datetime) -> datetime: + """ + Round a datetime up to the next minute if it has seconds/microseconds. + + Args: + dt: The datetime to round. + + Returns: + A datetime rounded up to the next minute. + """ + if dt.second > 0 or dt.microsecond > 0: + return normalize_datetime(dt) + timedelta(minutes=1) + return dt + + +def is_valid_day_in_month(year: int, month: int, day: int) -> bool: + """ + Check if a day is valid for a given month and year. + + Args: + year: The year. + month: The month (1-12). + day: The day to check. + + Returns: + True if the day is valid for the month, False otherwise. + """ + if month < 1 or month > 12: + return False + if day < 1: + return False + + max_day = get_days_in_month(year, month) + return day <= max_day + + +def increment_month(year: int, month: int) -> Tuple[int, int]: + """ + Increment month by 1, handling year rollover. + + Args: + year: The current year. + month: The current month (1-12). + + Returns: + A tuple of (new_year, new_month). + """ + month += 1 + if month > 12: + month = 1 + year += 1 + return year, month + + +def decrement_month(year: int, month: int) -> Tuple[int, int]: + """ + Decrement month by 1, handling year rollover. + + Args: + year: The current year. + month: The current month (1-12). + + Returns: + A tuple of (new_year, new_month). + """ + month -= 1 + if month < 1: + month = 12 + year -= 1 + return year, month + + +def get_month_bounds(dt: datetime) -> Tuple[datetime, datetime]: + """ + Get the first and last moments of the month for a datetime. + + Args: + dt: The datetime to check. + + Returns: + A tuple of (first_moment, last_moment) of the month. + """ + first_day = dt.replace(day=1, hour=0, minute=0, second=0, microsecond=0) + + # Get last day of month + last_day_num = get_days_in_month(dt.year, dt.month) + last_day = dt.replace( + day=last_day_num, + hour=23, + minute=59, + second=59, + microsecond=999999 + ) + + return first_day, last_day + + +def get_week_bounds(dt: datetime) -> Tuple[datetime, datetime]: + """ + Get the first and last moments of the week for a datetime. + Week starts on Sunday in cron. + + Args: + dt: The datetime to check. + + Returns: + A tuple of (first_moment, last_moment) of the week. + """ + # Get current weekday in cron format + cron_weekday = get_weekday(dt) + + # Calculate days to Sunday (start of week) + days_to_sunday = cron_weekday + + # Get Sunday at 00:00 + sunday = dt - timedelta(days=days_to_sunday) + sunday = sunday.replace(hour=0, minute=0, second=0, microsecond=0) + + # Get Saturday at 23:59:59.999999 + saturday = sunday + timedelta(days=6) + saturday = saturday.replace(hour=23, minute=59, second=59, microsecond=999999) + + return sunday, saturday \ No newline at end of file diff --git a/tests/test_time_utils.py b/tests/test_time_utils.py new file mode 100644 index 0000000..b4f88d4 --- /dev/null +++ b/tests/test_time_utils.py @@ -0,0 +1,279 @@ +"""Tests for time calculation utilities.""" + +import sys +from datetime import datetime +from pathlib import Path + +import pytest + +# Add src to path for testing +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from cronpal.time_utils import ( + decrement_month, + get_days_in_month, + get_month_bounds, + get_month_day_count, + get_next_day, + get_next_hour, + get_next_minute, + get_next_month, + get_next_year, + get_week_bounds, + get_weekday, + increment_month, + is_leap_year, + is_valid_day_in_month, + normalize_datetime, + round_to_next_minute, +) + + +class TestTimeIncrements: + """Tests for time increment functions.""" + + def test_get_next_minute(self): + """Test getting next minute.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + next_dt = get_next_minute(dt) + assert next_dt == datetime(2024, 1, 15, 10, 31, 45) + + def test_get_next_minute_hour_rollover(self): + """Test getting next minute with hour rollover.""" + dt = datetime(2024, 1, 15, 10, 59, 0) + next_dt = get_next_minute(dt) + assert next_dt == datetime(2024, 1, 15, 11, 0, 0) + + def test_get_next_minute_day_rollover(self): + """Test getting next minute with day rollover.""" + dt = datetime(2024, 1, 15, 23, 59, 0) + next_dt = get_next_minute(dt) + assert next_dt == datetime(2024, 1, 16, 0, 0, 0) + + def test_get_next_hour(self): + """Test getting next hour.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + next_dt = get_next_hour(dt) + assert next_dt == datetime(2024, 1, 15, 11, 0, 0) + + def test_get_next_hour_day_rollover(self): + """Test getting next hour with day rollover.""" + dt = datetime(2024, 1, 15, 23, 30, 0) + next_dt = get_next_hour(dt) + assert next_dt == datetime(2024, 1, 16, 0, 0, 0) + + def test_get_next_day(self): + """Test getting next day.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + next_dt = get_next_day(dt) + assert next_dt == datetime(2024, 1, 16, 0, 0, 0) + + def test_get_next_day_month_rollover(self): + """Test getting next day with month rollover.""" + dt = datetime(2024, 1, 31, 10, 30, 0) + next_dt = get_next_day(dt) + assert next_dt == datetime(2024, 2, 1, 0, 0, 0) + + def test_get_next_day_year_rollover(self): + """Test getting next day with year rollover.""" + dt = datetime(2024, 12, 31, 10, 30, 0) + next_dt = get_next_day(dt) + assert next_dt == datetime(2025, 1, 1, 0, 0, 0) + + def test_get_next_month(self): + """Test getting next month.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + next_dt = get_next_month(dt) + assert next_dt == datetime(2024, 2, 1, 0, 0, 0) + + def test_get_next_month_year_rollover(self): + """Test getting next month with year rollover.""" + dt = datetime(2024, 12, 15, 10, 30, 0) + next_dt = get_next_month(dt) + assert next_dt == datetime(2025, 1, 1, 0, 0, 0) + + def test_get_next_year(self): + """Test getting next year.""" + dt = datetime(2024, 6, 15, 10, 30, 45) + next_dt = get_next_year(dt) + assert next_dt == datetime(2025, 1, 1, 0, 0, 0) + + +class TestCalendarFunctions: + """Tests for calendar-related functions.""" + + def test_get_days_in_month_january(self): + """Test days in January.""" + assert get_days_in_month(2024, 1) == 31 + + def test_get_days_in_month_february_non_leap(self): + """Test days in February (non-leap year).""" + assert get_days_in_month(2023, 2) == 28 + + def test_get_days_in_month_february_leap(self): + """Test days in February (leap year).""" + assert get_days_in_month(2024, 2) == 29 + + def test_get_days_in_month_april(self): + """Test days in April.""" + assert get_days_in_month(2024, 4) == 30 + + def test_is_leap_year_true(self): + """Test leap year detection for leap years.""" + assert is_leap_year(2024) is True + assert is_leap_year(2000) is True + assert is_leap_year(2020) is True + + def test_is_leap_year_false(self): + """Test leap year detection for non-leap years.""" + assert is_leap_year(2023) is False + assert is_leap_year(1900) is False + assert is_leap_year(2100) is False + + def test_get_weekday_sunday(self): + """Test weekday for Sunday.""" + dt = datetime(2024, 1, 7) # Sunday + assert get_weekday(dt) == 0 + + def test_get_weekday_monday(self): + """Test weekday for Monday.""" + dt = datetime(2024, 1, 8) # Monday + assert get_weekday(dt) == 1 + + def test_get_weekday_saturday(self): + """Test weekday for Saturday.""" + dt = datetime(2024, 1, 6) # Saturday + assert get_weekday(dt) == 6 + + def test_get_month_day_count(self): + """Test getting day count for current month.""" + dt = datetime(2024, 2, 15) # February in leap year + assert get_month_day_count(dt) == 29 + + +class TestDateTimeNormalization: + """Tests for datetime normalization functions.""" + + def test_normalize_datetime(self): + """Test datetime normalization.""" + dt = datetime(2024, 1, 15, 10, 30, 45, 123456) + normalized = normalize_datetime(dt) + assert normalized == datetime(2024, 1, 15, 10, 30, 0, 0) + + def test_round_to_next_minute_no_rounding(self): + """Test rounding when no rounding is needed.""" + dt = datetime(2024, 1, 15, 10, 30, 0, 0) + rounded = round_to_next_minute(dt) + assert rounded == dt + + def test_round_to_next_minute_with_seconds(self): + """Test rounding with seconds.""" + dt = datetime(2024, 1, 15, 10, 30, 45, 0) + rounded = round_to_next_minute(dt) + assert rounded == datetime(2024, 1, 15, 10, 31, 0, 0) + + def test_round_to_next_minute_with_microseconds(self): + """Test rounding with microseconds.""" + dt = datetime(2024, 1, 15, 10, 30, 0, 123456) + rounded = round_to_next_minute(dt) + assert rounded == datetime(2024, 1, 15, 10, 31, 0, 0) + + +class TestValidation: + """Tests for validation functions.""" + + def test_is_valid_day_in_month_valid(self): + """Test valid days in month.""" + assert is_valid_day_in_month(2024, 1, 1) is True + assert is_valid_day_in_month(2024, 1, 31) is True + assert is_valid_day_in_month(2024, 2, 29) is True # Leap year + assert is_valid_day_in_month(2024, 4, 30) is True + + def test_is_valid_day_in_month_invalid(self): + """Test invalid days in month.""" + assert is_valid_day_in_month(2024, 1, 0) is False + assert is_valid_day_in_month(2024, 1, 32) is False + assert is_valid_day_in_month(2023, 2, 29) is False # Non-leap year + assert is_valid_day_in_month(2024, 4, 31) is False + + def test_is_valid_day_in_month_invalid_month(self): + """Test invalid month.""" + assert is_valid_day_in_month(2024, 0, 15) is False + assert is_valid_day_in_month(2024, 13, 15) is False + + +class TestMonthOperations: + """Tests for month increment/decrement operations.""" + + def test_increment_month_normal(self): + """Test incrementing month normally.""" + year, month = increment_month(2024, 5) + assert year == 2024 + assert month == 6 + + def test_increment_month_year_rollover(self): + """Test incrementing month with year rollover.""" + year, month = increment_month(2024, 12) + assert year == 2025 + assert month == 1 + + def test_decrement_month_normal(self): + """Test decrementing month normally.""" + year, month = decrement_month(2024, 5) + assert year == 2024 + assert month == 4 + + def test_decrement_month_year_rollover(self): + """Test decrementing month with year rollover.""" + year, month = decrement_month(2024, 1) + assert year == 2023 + assert month == 12 + + +class TestBounds: + """Tests for getting time period bounds.""" + + def test_get_month_bounds(self): + """Test getting month boundaries.""" + dt = datetime(2024, 2, 15, 10, 30, 45) + first, last = get_month_bounds(dt) + + assert first == datetime(2024, 2, 1, 0, 0, 0, 0) + assert last == datetime(2024, 2, 29, 23, 59, 59, 999999) + + def test_get_month_bounds_january(self): + """Test getting January boundaries.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + first, last = get_month_bounds(dt) + + assert first == datetime(2024, 1, 1, 0, 0, 0, 0) + assert last == datetime(2024, 1, 31, 23, 59, 59, 999999) + + def test_get_week_bounds_midweek(self): + """Test getting week boundaries from midweek.""" + # Wednesday, January 10, 2024 + dt = datetime(2024, 1, 10, 10, 30, 45) + first, last = get_week_bounds(dt) + + # Should be Sunday, January 7 to Saturday, January 13 + assert first == datetime(2024, 1, 7, 0, 0, 0, 0) + assert last == datetime(2024, 1, 13, 23, 59, 59, 999999) + + def test_get_week_bounds_sunday(self): + """Test getting week boundaries from Sunday.""" + # Sunday, January 7, 2024 + dt = datetime(2024, 1, 7, 10, 30, 45) + first, last = get_week_bounds(dt) + + assert first == datetime(2024, 1, 7, 0, 0, 0, 0) + assert last == datetime(2024, 1, 13, 23, 59, 59, 999999) + + def test_get_week_bounds_saturday(self): + """Test getting week boundaries from Saturday.""" + # Saturday, January 6, 2024 + dt = datetime(2024, 1, 6, 10, 30, 45) + first, last = get_week_bounds(dt) + + # Should be Sunday, December 31, 2023 to Saturday, January 6, 2024 + assert first == datetime(2023, 12, 31, 0, 0, 0, 0) + assert last == datetime(2024, 1, 6, 23, 59, 59, 999999) \ No newline at end of file