From b89457a188726b996fee8354b063912b612577aa Mon Sep 17 00:00:00 2001 From: Valentin Todorov Date: Mon, 22 Sep 2025 17:31:22 +0300 Subject: [PATCH] feat: add previous run time calculator --- src/cronpal/cli.py | 65 +++++- src/cronpal/parser.py | 7 + src/cronpal/scheduler.py | 235 ++++++++++++++++++- src/cronpal/time_utils.py | 90 ++++++++ tests/test_cli_previous.py | 343 ++++++++++++++++++++++++++++ tests/test_scheduler_previous.py | 366 ++++++++++++++++++++++++++++++ tests/test_time_utils_previous.py | 198 ++++++++++++++++ 7 files changed, 1301 insertions(+), 3 deletions(-) create mode 100644 tests/test_cli_previous.py create mode 100644 tests/test_scheduler_previous.py create mode 100644 tests/test_time_utils_previous.py diff --git a/src/cronpal/cli.py b/src/cronpal/cli.py index 564f745..43f5ef4 100644 --- a/src/cronpal/cli.py +++ b/src/cronpal/cli.py @@ -40,7 +40,7 @@ def main(args=None): # Parse as special string cron_expr = special_parser.parse(parsed_args.expression) - print(f"✓ Valid cron expression: {cron_expr}") + print(f"✔ Valid cron expression: {cron_expr}") if parsed_args.verbose: print(f" Special string: {cron_expr.raw_expression}") @@ -65,7 +65,7 @@ def main(args=None): cron_expr.month = field_parser.parse_month(fields[3]) cron_expr.day_of_week = field_parser.parse_day_of_week(fields[4]) - print(f"✓ Valid cron expression: {cron_expr}") + print(f"✔ Valid cron expression: {cron_expr}") if parsed_args.verbose: print(f" Raw expression: {cron_expr.raw_expression}") @@ -76,6 +76,10 @@ def main(args=None): if parsed_args.next is not None: _print_next_runs(cron_expr, parsed_args.next) + # Show previous run times if requested + if parsed_args.previous is not None: + _print_previous_runs(cron_expr, parsed_args.previous) + return 0 except CronPalError as e: @@ -242,5 +246,62 @@ def _print_next_runs(cron_expr: CronExpression, count: int): print(f"\nNext runs: Error calculating - {e}") +def _print_previous_runs(cron_expr: CronExpression, count: int): + """ + Print the previous run times for a cron expression. + + Args: + cron_expr: The CronExpression to calculate runs for. + count: Number of previous runs to show. + """ + # Don't show previous runs for @reboot + if cron_expr.raw_expression.lower() == "@reboot": + print("\nPrevious runs: @reboot only runs at system startup") + return + + # Make sure we have parsed fields + if not cron_expr.is_valid(): + print("\nPrevious runs: Cannot calculate - incomplete expression") + return + + try: + scheduler = CronScheduler(cron_expr) + previous_runs = scheduler.get_previous_runs(count) + + print(f"\nPrevious {count} run{'s' if count != 1 else ''} (most recent first):") + for i, run_time in enumerate(previous_runs, 1): + # Format the datetime nicely + formatted = run_time.strftime("%Y-%m-%d %H:%M:%S %A") + + # Add relative time for first few entries + if i <= 3: + now = datetime.now() + delta = now - run_time + + if delta.days == 0: + if delta.seconds < 3600: + minutes = delta.seconds // 60 + relative = f"{minutes} minute{'s' if minutes != 1 else ''} ago" + else: + hours = delta.seconds // 3600 + relative = f"{hours} hour{'s' if hours != 1 else ''} ago" + elif delta.days == 1: + relative = "yesterday" + elif delta.days < 7: + relative = f"{delta.days} days ago" + else: + relative = "" + + if relative: + print(f" {i}. {formatted} ({relative})") + else: + print(f" {i}. {formatted}") + else: + print(f" {i}. {formatted}") + + except Exception as e: + print(f"\nPrevious runs: Error calculating - {e}") + + if __name__ == "__main__": sys.exit(main()) \ No newline at end of file diff --git a/src/cronpal/parser.py b/src/cronpal/parser.py index 74c4955..d9e6191 100644 --- a/src/cronpal/parser.py +++ b/src/cronpal/parser.py @@ -56,4 +56,11 @@ def create_parser(): help="Show next N execution times (default: 5)" ) + parser.add_argument( + "-p", "--previous", + type=int, + metavar="N", + help="Show previous N execution times" + ) + return parser \ No newline at end of file diff --git a/src/cronpal/scheduler.py b/src/cronpal/scheduler.py index 0acc0db..f24ed0a 100644 --- a/src/cronpal/scheduler.py +++ b/src/cronpal/scheduler.py @@ -10,9 +10,14 @@ get_next_hour, get_next_minute, get_next_month, + get_previous_day, + get_previous_hour, + get_previous_minute, + get_previous_month, get_weekday, is_valid_day_in_month, round_to_next_minute, + round_to_previous_minute, ) @@ -99,6 +104,71 @@ def get_next_runs(self, count: int, after: Optional[datetime] = None) -> List[da return runs + def get_previous_run(self, before: Optional[datetime] = None) -> datetime: + """ + Calculate the previous run time for the cron expression. + + Args: + before: The datetime to start searching from. + Defaults to current time if not provided. + + Returns: + The previous datetime when the cron expression ran. + """ + if before is None: + before = datetime.now() + + # Round down to previous minute if needed + current = round_to_previous_minute(before) + + # Maximum iterations to prevent infinite loops + max_iterations = 10000 + iterations = 0 + + while iterations < max_iterations: + iterations += 1 + + # Check if current time matches the cron expression + if self._matches_time(current): + return current + + # Move to previous possible time + current = self._retreat_to_previous_possible(current) + + raise CronPalError("Could not find previous run time within reasonable limits") + + def get_previous_runs(self, count: int, before: Optional[datetime] = None) -> List[datetime]: + """ + Calculate multiple previous run times for the cron expression. + + Args: + count: Number of previous run times to calculate. + before: The datetime to start searching from. + Defaults to current time if not provided. + + Returns: + List of previous run times (most recent first). + + Raises: + ValueError: If count is less than 1. + """ + if count < 1: + raise ValueError("Count must be at least 1") + + if before is None: + before = datetime.now() + + runs = [] + current = before + + for _ in range(count): + previous_run = self.get_previous_run(current) + runs.append(previous_run) + # Start next search 1 minute before the found time + current = previous_run - timedelta(minutes=1) + + return runs + def _matches_time(self, dt: datetime) -> bool: """ Check if a datetime matches the cron expression. @@ -166,6 +236,34 @@ def _advance_to_next_possible(self, dt: datetime) -> datetime: # If no valid day this month, try next month return self._get_next_month(dt) + def _retreat_to_previous_possible(self, dt: datetime) -> datetime: + """ + Retreat datetime to the previous possible matching time. + + Args: + dt: The current datetime. + + Returns: + The previous datetime that could potentially match. + """ + # Try to retreat minute first + prev_minute = self._get_previous_minute(dt) + if prev_minute is not None: + return prev_minute + + # If no valid minute in this hour, try previous hour + prev_hour = self._get_previous_hour(dt) + if prev_hour is not None: + return prev_hour + + # If no valid hour today, try previous day + prev_day = self._get_previous_day(dt) + if prev_day is not None: + return prev_day + + # If no valid day this month, try previous month + return self._get_previous_month(dt) + def _get_next_minute(self, dt: datetime) -> Optional[datetime]: """ Get the next valid minute after the current time. @@ -185,6 +283,25 @@ def _get_next_minute(self, dt: datetime) -> Optional[datetime]: return None + def _get_previous_minute(self, dt: datetime) -> Optional[datetime]: + """ + Get the previous valid minute before the current time. + + Args: + dt: The current datetime. + + Returns: + Previous valid minute in the same hour, or None if no valid minute. + """ + current_minute = dt.minute + valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True) + + for minute in valid_minutes: + if minute < current_minute: + return dt.replace(minute=minute, second=0, microsecond=0) + + return None + def _get_next_hour(self, dt: datetime) -> Optional[datetime]: """ Get the next valid hour after the current time. @@ -208,6 +325,29 @@ def _get_next_hour(self, dt: datetime) -> Optional[datetime]: return None + def _get_previous_hour(self, dt: datetime) -> Optional[datetime]: + """ + Get the previous valid hour before the current time. + + Args: + dt: The current datetime. + + Returns: + Previous valid hour in the same day, or None if no valid hour. + """ + current_hour = dt.hour + valid_hours = sorted(self.cron_expr.hour.parsed_values, reverse=True) + valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True) + + # Last minute of the previous valid hour + last_minute = valid_minutes[0] if valid_minutes else 59 + + for hour in valid_hours: + if hour < current_hour: + return dt.replace(hour=hour, minute=last_minute, second=0, microsecond=0) + + return None + def _get_next_day(self, dt: datetime) -> Optional[datetime]: """ Get the next valid day after the current time. @@ -248,6 +388,43 @@ def _get_next_day(self, dt: datetime) -> Optional[datetime]: return None + def _get_previous_day(self, dt: datetime) -> Optional[datetime]: + """ + Get the previous valid day before the current time. + + Args: + dt: The current datetime. + + Returns: + Previous valid day in the same month, or None if no valid day. + """ + current_day = dt.day + valid_hours = sorted(self.cron_expr.hour.parsed_values, reverse=True) + valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True) + + # Last time of the day + last_hour = valid_hours[0] if valid_hours else 23 + last_minute = valid_minutes[0] if valid_minutes else 59 + + # Check previous days in the month + for day in range(current_day - 1, 0, -1): + test_dt = dt.replace(day=day, hour=last_hour, minute=last_minute, + second=0, microsecond=0) + + # Check if this day matches day constraints + day_of_month_match = day in self.cron_expr.day_of_month.parsed_values + day_of_week_match = get_weekday(test_dt) in self.cron_expr.day_of_week.parsed_values + + # Apply OR logic for day fields if both are restricted + if not self.cron_expr.day_of_month.is_wildcard() and not self.cron_expr.day_of_week.is_wildcard(): + if day_of_month_match or day_of_week_match: + return test_dt + else: + if day_of_month_match and day_of_week_match: + return test_dt + + return None + def _get_next_month(self, dt: datetime) -> datetime: """ Get the first valid time in the next valid month. @@ -302,4 +479,60 @@ def _get_next_month(self, dt: datetime) -> datetime: return test_dt # This should rarely happen unless the cron expression is very restrictive - raise CronPalError("Could not find valid next month") \ No newline at end of file + raise CronPalError("Could not find valid next month") + + def _get_previous_month(self, dt: datetime) -> datetime: + """ + Get the last valid time in the previous valid month. + + Args: + dt: The current datetime. + + Returns: + Last valid time in the previous valid month. + """ + valid_months = sorted(self.cron_expr.month.parsed_values, reverse=True) + valid_hours = sorted(self.cron_expr.hour.parsed_values, reverse=True) + valid_minutes = sorted(self.cron_expr.minute.parsed_values, reverse=True) + + # Last time of any day + last_hour = valid_hours[0] if valid_hours else 23 + last_minute = valid_minutes[0] if valid_minutes else 59 + + # Start searching from current year + current_year = dt.year + current_month = dt.month + + # Search for up to 10 years back + for year_offset in range(10): + search_year = current_year - year_offset + + # Determine which months to check this year + if year_offset == 0: + # For current year, only check months before current month + months_to_check = [m for m in valid_months if m < current_month] + else: + # For past years, check all valid months + months_to_check = valid_months + + for month in months_to_check: + # Find last valid day in this month + for day in range(31, 0, -1): + if not is_valid_day_in_month(search_year, month, day): + continue + + test_dt = datetime(search_year, month, day, last_hour, last_minute, 0, 0) + + # Check day constraints + day_of_month_match = day in self.cron_expr.day_of_month.parsed_values + day_of_week_match = get_weekday(test_dt) in self.cron_expr.day_of_week.parsed_values + + if not self.cron_expr.day_of_month.is_wildcard() and not self.cron_expr.day_of_week.is_wildcard(): + if day_of_month_match or day_of_week_match: + return test_dt + else: + if day_of_month_match and day_of_week_match: + return test_dt + + # This should rarely happen unless the cron expression is very restrictive + raise CronPalError("Could not find valid previous month") \ No newline at end of file diff --git a/src/cronpal/time_utils.py b/src/cronpal/time_utils.py index 5f7bd98..2c924eb 100644 --- a/src/cronpal/time_utils.py +++ b/src/cronpal/time_utils.py @@ -18,6 +18,19 @@ def get_next_minute(dt: datetime) -> datetime: return dt + timedelta(minutes=1) +def get_previous_minute(dt: datetime) -> datetime: + """ + Get the previous minute from the given datetime. + + Args: + dt: The datetime to decrement. + + Returns: + A datetime object representing the previous 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). @@ -32,6 +45,20 @@ def get_next_hour(dt: datetime) -> datetime: return next_hour + timedelta(hours=1) +def get_previous_hour(dt: datetime) -> datetime: + """ + Get the previous hour from the given datetime (minute set to 59). + + Args: + dt: The datetime to decrement. + + Returns: + A datetime object at the end of the previous hour. + """ + prev_hour = dt.replace(minute=59, second=0, microsecond=0) + return prev_hour - timedelta(hours=1) + + def get_next_day(dt: datetime) -> datetime: """ Get the next day from the given datetime (time set to 00:00). @@ -46,6 +73,20 @@ def get_next_day(dt: datetime) -> datetime: return next_day + timedelta(days=1) +def get_previous_day(dt: datetime) -> datetime: + """ + Get the previous day from the given datetime (time set to 23:59). + + Args: + dt: The datetime to decrement. + + Returns: + A datetime object at the end of the previous day. + """ + prev_day = dt.replace(hour=23, minute=59, second=0, microsecond=0) + return prev_day - timedelta(days=1) + + def get_next_month(dt: datetime) -> datetime: """ Get the first day of the next month. @@ -67,6 +108,29 @@ def get_next_month(dt: datetime) -> datetime: return datetime(year, month, 1, 0, 0, 0, 0) +def get_previous_month(dt: datetime) -> datetime: + """ + Get the last day of the previous month. + + Args: + dt: The datetime to decrement. + + Returns: + A datetime object at the end of the previous month. + """ + # Calculate previous month + year = dt.year + month = dt.month - 1 + + if month < 1: + month = 12 + year -= 1 + + # Get last day of previous month + last_day = get_days_in_month(year, month) + return datetime(year, month, last_day, 23, 59, 0, 0) + + def get_next_year(dt: datetime) -> datetime: """ Get the first day of the next year. @@ -80,6 +144,19 @@ def get_next_year(dt: datetime) -> datetime: return datetime(dt.year + 1, 1, 1, 0, 0, 0, 0) +def get_previous_year(dt: datetime) -> datetime: + """ + Get the last day of the previous year. + + Args: + dt: The datetime to decrement. + + Returns: + A datetime object at the end of the previous year. + """ + return datetime(dt.year - 1, 12, 31, 23, 59, 0, 0) + + def get_days_in_month(year: int, month: int) -> int: """ Get the number of days in a given month. @@ -169,6 +246,19 @@ def round_to_next_minute(dt: datetime) -> datetime: return dt +def round_to_previous_minute(dt: datetime) -> datetime: + """ + Round a datetime down to the previous minute, removing seconds/microseconds. + + Args: + dt: The datetime to round. + + Returns: + A datetime rounded down to the minute. + """ + return dt.replace(second=0, microsecond=0) + + 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. diff --git a/tests/test_cli_previous.py b/tests/test_cli_previous.py new file mode 100644 index 0000000..be46a3c --- /dev/null +++ b/tests/test_cli_previous.py @@ -0,0 +1,343 @@ +"""Tests for CLI previous runs functionality.""" + +import subprocess +import sys +from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import patch + +import pytest + +# Add src to path for testing +sys.path.insert(0, str(Path(__file__).parent.parent / "src")) + +from cronpal.cli import main + + +class TestCLIPreviousRuns: + """Tests for CLI --previous flag functionality.""" + + def test_previous_flag_single(self): + """Test --previous flag with single run.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 * * *", "--previous", "1"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 1 run" in output + assert "1." in output + + def test_previous_flag_multiple(self): + """Test --previous flag with multiple runs.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 * * *", "--previous", "3"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 3 runs" in output + assert "most recent first" in output + assert "1." in output + assert "2." in output + assert "3." in output + + @patch('cronpal.scheduler.datetime') + def test_previous_runs_with_fixed_time(self, mock_datetime): + """Test previous runs with fixed current time.""" + import io + import contextlib + + # Fix the current time for predictable output + fixed_time = datetime(2024, 1, 15, 14, 30, 0) + mock_datetime.now.return_value = fixed_time + mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs) + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 12 * * *", "--previous", "2"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 2 runs" in output + # Should show 12:00 today and yesterday + assert "2024-01-15 12:00:00" in output + assert "2024-01-14 12:00:00" in output + + def test_previous_runs_every_minute(self): + """Test previous runs for every minute.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["* * * * *", "--previous", "3"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 3 runs" in output + # Should have relative time descriptions + assert "ago" in output.lower() or "minute" in output.lower() + + def test_previous_runs_with_verbose(self): + """Test previous runs with verbose flag.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 * * *", "--previous", "2", "--verbose"]) + + output = f.getvalue() + assert result == 0 + # Should show field details AND previous runs + assert "Minute field:" in output + assert "Hour field:" in output + assert "Previous 2 runs" in output + + def test_previous_runs_special_string(self): + """Test previous runs with special string.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["@daily", "--previous", "3"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 3 runs" in output + # Daily should show midnight times + assert "00:00:00" in output + + def test_previous_runs_special_string_reboot(self): + """Test previous runs with @reboot.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["@reboot", "--previous", "5"]) + + output = f.getvalue() + assert result == 0 + assert "@reboot only runs at system startup" in output + + def test_previous_runs_hourly(self): + """Test previous runs for hourly schedule.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 * * * *", "--previous", "3"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 3 runs" in output + # All should be at minute 00 + assert output.count(":00:00") >= 3 + + def test_previous_runs_specific_weekday(self): + """Test previous runs for specific weekday.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 * * MON", "--previous", "3"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 3 runs" in output + # All should be Mondays + assert output.count("Monday") == 3 + + def test_previous_runs_complex_expression(self): + """Test previous runs for complex expression.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["*/15 9-17 * * MON-FRI", "--previous", "5"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 5 runs" in output + # Should not include weekends + assert "Saturday" not in output + assert "Sunday" not in output + + def test_previous_runs_with_month_names(self): + """Test previous runs with month names.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 1 JAN,JUL *", "--previous", "2"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 2 runs" in output + + def test_previous_runs_with_day_names(self): + """Test previous runs with day names.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 * * SUN,WED", "--previous", "4"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 4 runs" in output + # Should include Sundays and Wednesdays + assert "Sunday" in output or "Wednesday" in output + + def test_previous_runs_with_step_values(self): + """Test previous runs with step values.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["*/10 * * * *", "--previous", "3"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 3 runs" in output + # Minutes should be 00, 10, 20, 30, 40, or 50 + + def test_previous_runs_invalid_expression(self): + """Test previous runs with invalid expression.""" + import io + import contextlib + + f_err = io.StringIO() + with contextlib.redirect_stderr(f_err): + result = main(["invalid", "--previous", "5"]) + + error_output = f_err.getvalue() + assert result == 1 + # Should show error, not previous runs + assert "Invalid" in error_output + + def test_previous_runs_relative_time_today(self): + """Test that relative time shows for today.""" + import io + import contextlib + + # Use an expression that ran recently + now = datetime.now() + prev_hour = (now.hour - 1) % 24 + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main([f"0 * * * *", "--previous", "1"]) + + output = f.getvalue() + assert result == 0 + # Should show relative time if it's today + assert "ago" in output.lower() + + def test_previous_runs_formatting(self): + """Test previous runs output formatting.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 * * *", "--previous", "5"]) + + output = f.getvalue() + assert result == 0 + # Check formatting + assert "Previous 5 runs" in output + + # Check numbering + for i in range(1, 6): + assert f"{i}." in output + + # Check date format (YYYY-MM-DD HH:MM:SS Weekday) + assert "2" in output # Year starts with 2 + assert ":" in output # Time separator + + def test_both_next_and_previous(self): + """Test using both --next and --previous flags.""" + import io + import contextlib + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 * * *", "--next", "2", "--previous", "2"]) + + output = f.getvalue() + assert result == 0 + # Should show both + assert "Next 2 runs" in output + assert "Previous 2 runs" in output + + def test_previous_leap_year(self): + """Test previous runs including February 29.""" + import io + import contextlib + + # Use a date after Feb 29, 2024 (leap year) + with patch('cronpal.scheduler.datetime') as mock_datetime: + fixed_time = datetime(2024, 3, 1, 0, 0, 0) + mock_datetime.now.return_value = fixed_time + mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs) + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 0 29 2 *", "--previous", "2"]) + + output = f.getvalue() + assert result == 0 + assert "Previous 2 runs" in output + # Should show 2024-02-29 and 2020-02-29 + assert "2024-02-29" in output + + def test_previous_runs_yesterday(self): + """Test that 'yesterday' appears in relative time.""" + import io + import contextlib + + # Use an expression that runs daily + with patch('cronpal.scheduler.datetime') as mock_datetime: + fixed_time = datetime(2024, 1, 15, 14, 0, 0) + mock_datetime.now.return_value = fixed_time + mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs) + + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 12 * * *", "--previous", "1"]) + + output = f.getvalue() + assert result == 0 + # Should show yesterday for a run from previous day + # Or hours ago for today's run + + def test_previous_runs_hours_ago(self): + """Test that 'hours ago' appears for recent runs.""" + import io + import contextlib + + # Use an expression that runs hourly + f = io.StringIO() + with contextlib.redirect_stdout(f): + result = main(["0 * * * *", "--previous", "1"]) + + output = f.getvalue() + assert result == 0 + # Should show relative time + assert "ago" in output.lower() \ No newline at end of file diff --git a/tests/test_scheduler_previous.py b/tests/test_scheduler_previous.py new file mode 100644 index 0000000..9867580 --- /dev/null +++ b/tests/test_scheduler_previous.py @@ -0,0 +1,366 @@ +"""Tests for the cron scheduler previous run functionality.""" + +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.exceptions import CronPalError +from cronpal.field_parser import FieldParser +from cronpal.models import CronExpression +from cronpal.scheduler import CronScheduler + + +def create_cron_expression(expr_str: str) -> CronExpression: + """Helper to create a parsed CronExpression.""" + fields = expr_str.split() + expr = CronExpression(expr_str) + parser = FieldParser() + + expr.minute = parser.parse_minute(fields[0]) + expr.hour = parser.parse_hour(fields[1]) + expr.day_of_month = parser.parse_day_of_month(fields[2]) + expr.month = parser.parse_month(fields[3]) + expr.day_of_week = parser.parse_day_of_week(fields[4]) + + return expr + + +class TestCronSchedulerPrevious: + """Tests for CronScheduler previous run methods.""" + + def test_every_minute_previous(self): + """Test previous run for every minute.""" + expr = create_cron_expression("* * * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 30, 0) + prev_run = scheduler.get_previous_run(start) + + # Should be the same time (already on a minute boundary) + assert prev_run == start + + def test_every_minute_with_seconds_previous(self): + """Test every minute with seconds in start time.""" + expr = create_cron_expression("* * * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 30, 45) + prev_run = scheduler.get_previous_run(start) + + # Should round down to same minute + assert prev_run == datetime(2024, 1, 15, 10, 30, 0) + + def test_specific_minute_previous(self): + """Test specific minute of every hour.""" + expr = create_cron_expression("15 * * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 30, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 15, 10, 15, 0) + + def test_specific_minute_previous_hour(self): + """Test specific minute when current minute is before.""" + expr = create_cron_expression("30 * * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 15, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 15, 9, 30, 0) + + def test_specific_hour_previous(self): + """Test specific hour of every day.""" + expr = create_cron_expression("0 14 * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 16, 0, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 15, 14, 0, 0) + + def test_specific_hour_previous_day(self): + """Test specific hour when current hour is before.""" + expr = create_cron_expression("0 18 * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 0, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 14, 18, 0, 0) + + def test_specific_day_of_month_previous(self): + """Test specific day of month.""" + expr = create_cron_expression("0 0 15 * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 20, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 15, 0, 0, 0) + + def test_specific_day_of_month_previous_month(self): + """Test specific day when current day is before.""" + expr = create_cron_expression("0 0 20 * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 2, 10, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 20, 0, 0, 0) + + def test_specific_month_previous(self): + """Test specific month.""" + expr = create_cron_expression("0 0 1 6 *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 8, 1, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 6, 1, 0, 0, 0) + + def test_specific_month_previous_year(self): + """Test specific month when current month is before.""" + expr = create_cron_expression("0 0 1 8 *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 6, 1, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2023, 8, 1, 0, 0, 0) + + def test_specific_weekday_previous(self): + """Test specific day of week (Monday).""" + expr = create_cron_expression("0 0 * * 1") + scheduler = CronScheduler(expr) + + # Start on Tuesday, January 16, 2024 + start = datetime(2024, 1, 16, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # Previous Monday is January 15 + assert prev_run == datetime(2024, 1, 15, 0, 0, 0) + assert prev_run.weekday() == 0 # Monday in Python + + def test_weekday_and_monthday_previous(self): + """Test expression with both day of month and day of week.""" + # Run on 15th OR Mondays (OR logic) + expr = create_cron_expression("0 0 15 * 1") + scheduler = CronScheduler(expr) + + # Start on January 16 (Tuesday) + start = datetime(2024, 1, 16, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # Previous should be January 15 (Monday) - matches both + assert prev_run == datetime(2024, 1, 15, 0, 0, 0) + + def test_every_15_minutes_previous(self): + """Test every 15 minutes.""" + expr = create_cron_expression("*/15 * * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 20, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 15, 10, 15, 0) + + def test_business_hours_previous(self): + """Test business hours (9-17 on weekdays).""" + expr = create_cron_expression("0 9-17 * * 1-5") + scheduler = CronScheduler(expr) + + # Start Friday afternoon + start = datetime(2024, 1, 12, 16, 30, 0) # Friday + prev_run = scheduler.get_previous_run(start) + + # Previous is 4 PM same day + assert prev_run == datetime(2024, 1, 12, 16, 0, 0) + + def test_business_hours_weekend_skip_previous(self): + """Test business hours skipping weekend.""" + expr = create_cron_expression("0 17 * * 1-5") + scheduler = CronScheduler(expr) + + # Start Monday morning + start = datetime(2024, 1, 15, 8, 0, 0) # Monday morning + prev_run = scheduler.get_previous_run(start) + + # Previous is Friday evening + assert prev_run == datetime(2024, 1, 12, 17, 0, 0) + assert prev_run.weekday() == 4 # Friday + + def test_last_day_of_month_previous(self): + """Test handling of last day of month.""" + expr = create_cron_expression("0 0 31 * *") + scheduler = CronScheduler(expr) + + # Start in April (no 31st) + start = datetime(2024, 4, 1, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # Should be March 31 + assert prev_run == datetime(2024, 3, 31, 0, 0, 0) + + def test_february_29_previous(self): + """Test February 29 in leap year.""" + expr = create_cron_expression("0 0 29 2 *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 3, 1, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # 2024 is a leap year + assert prev_run == datetime(2024, 2, 29, 0, 0, 0) + + def test_february_29_non_leap_year_previous(self): + """Test February 29 in non-leap year.""" + expr = create_cron_expression("0 0 29 2 *") + scheduler = CronScheduler(expr) + + start = datetime(2023, 3, 1, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # 2023 is not a leap year, go back to 2020 + assert prev_run == datetime(2020, 2, 29, 0, 0, 0) + + def test_complex_expression_previous(self): + """Test complex expression with multiple constraints.""" + expr = create_cron_expression("30 2 1-15 * MON-FRI") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 16, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # Should find a weekday in first half of month + assert prev_run.day <= 15 + assert prev_run.hour == 2 + assert prev_run.minute == 30 + assert prev_run.weekday() < 5 # Monday-Friday + + def test_range_with_step_previous(self): + """Test range with step values.""" + expr = create_cron_expression("0 */4 * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 30, 0) + prev_run = scheduler.get_previous_run(start) + + # Previous 4-hour interval is 8:00 + assert prev_run == datetime(2024, 1, 15, 8, 0, 0) + + def test_list_of_values_previous(self): + """Test list of specific values.""" + expr = create_cron_expression("0 9,12,15 * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 14, 0, 0) + prev_run = scheduler.get_previous_run(start) + + assert prev_run == datetime(2024, 1, 15, 12, 0, 0) + + def test_get_previous_runs_multiple(self): + """Test getting multiple previous runs.""" + expr = create_cron_expression("0 0 * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 0, 0) + prev_runs = scheduler.get_previous_runs(3, start) + + assert len(prev_runs) == 3 + # Should be today at midnight, yesterday, day before + assert prev_runs[0] == datetime(2024, 1, 15, 0, 0, 0) + assert prev_runs[1] == datetime(2024, 1, 14, 0, 0, 0) + assert prev_runs[2] == datetime(2024, 1, 13, 0, 0, 0) + + def test_get_previous_runs_hourly(self): + """Test getting multiple previous runs for hourly schedule.""" + expr = create_cron_expression("30 * * * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 10, 45, 0) + prev_runs = scheduler.get_previous_runs(3, start) + + assert len(prev_runs) == 3 + assert prev_runs[0] == datetime(2024, 1, 15, 10, 30, 0) + assert prev_runs[1] == datetime(2024, 1, 15, 9, 30, 0) + assert prev_runs[2] == datetime(2024, 1, 15, 8, 30, 0) + + def test_get_previous_runs_invalid_count(self): + """Test get_previous_runs with invalid count.""" + expr = create_cron_expression("0 0 * * *") + scheduler = CronScheduler(expr) + + with pytest.raises(ValueError, match="Count must be at least 1"): + scheduler.get_previous_runs(0) + + with pytest.raises(ValueError, match="Count must be at least 1"): + scheduler.get_previous_runs(-1) + + def test_sunday_as_zero_previous(self): + """Test Sunday as day 0.""" + expr = create_cron_expression("0 0 * * 0") + scheduler = CronScheduler(expr) + + # Start on Monday + start = datetime(2024, 1, 15, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # Previous Sunday is January 14 + assert prev_run == datetime(2024, 1, 14, 0, 0, 0) + assert prev_run.weekday() == 6 # Sunday in Python + + def test_sunday_as_seven_previous(self): + """Test Sunday as day 7 (should be same as 0).""" + expr = create_cron_expression("0 0 * * 7") + scheduler = CronScheduler(expr) + + # Start on Monday + start = datetime(2024, 1, 15, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # Previous Sunday is January 14 + assert prev_run == datetime(2024, 1, 14, 0, 0, 0) + assert prev_run.weekday() == 6 # Sunday in Python + + def test_minute_boundary_previous(self): + """Test that previous run respects minute boundaries.""" + expr = create_cron_expression("15 10 * * *") + scheduler = CronScheduler(expr) + + # Start just after the target time + start = datetime(2024, 1, 15, 10, 15, 30) + prev_run = scheduler.get_previous_run(start) + + # Should get the exact minute + assert prev_run == datetime(2024, 1, 15, 10, 15, 0) + + def test_year_boundary_previous(self): + """Test previous run crossing year boundary.""" + expr = create_cron_expression("0 0 31 12 *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 1, 15, 0, 0, 0) + prev_run = scheduler.get_previous_run(start) + + # Should be December 31 of previous year + assert prev_run == datetime(2023, 12, 31, 0, 0, 0) + + def test_monthly_on_specific_day_previous(self): + """Test monthly schedule on specific day.""" + expr = create_cron_expression("0 0 15 * *") + scheduler = CronScheduler(expr) + + start = datetime(2024, 3, 10, 0, 0, 0) + prev_runs = scheduler.get_previous_runs(3, start) + + assert len(prev_runs) == 3 + assert prev_runs[0] == datetime(2024, 2, 15, 0, 0, 0) + assert prev_runs[1] == datetime(2024, 1, 15, 0, 0, 0) + assert prev_runs[2] == datetime(2023, 12, 15, 0, 0, 0) \ No newline at end of file diff --git a/tests/test_time_utils_previous.py b/tests/test_time_utils_previous.py new file mode 100644 index 0000000..4ae2376 --- /dev/null +++ b/tests/test_time_utils_previous.py @@ -0,0 +1,198 @@ +"""Tests for previous 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 ( + get_previous_day, + get_previous_hour, + get_previous_minute, + get_previous_month, + get_previous_year, + round_to_previous_minute, +) + + +class TestPreviousTimeIncrements: + """Tests for previous time increment functions.""" + + def test_get_previous_minute(self): + """Test getting previous minute.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + prev_dt = get_previous_minute(dt) + assert prev_dt == datetime(2024, 1, 15, 10, 29, 45) + + def test_get_previous_minute_hour_rollover(self): + """Test getting previous minute with hour rollover.""" + dt = datetime(2024, 1, 15, 10, 0, 0) + prev_dt = get_previous_minute(dt) + assert prev_dt == datetime(2024, 1, 15, 9, 59, 0) + + def test_get_previous_minute_day_rollover(self): + """Test getting previous minute with day rollover.""" + dt = datetime(2024, 1, 15, 0, 0, 0) + prev_dt = get_previous_minute(dt) + assert prev_dt == datetime(2024, 1, 14, 23, 59, 0) + + def test_get_previous_hour(self): + """Test getting previous hour.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + prev_dt = get_previous_hour(dt) + assert prev_dt == datetime(2024, 1, 15, 9, 59, 0, 0) + + def test_get_previous_hour_day_rollover(self): + """Test getting previous hour with day rollover.""" + dt = datetime(2024, 1, 15, 0, 30, 0) + prev_dt = get_previous_hour(dt) + assert prev_dt == datetime(2024, 1, 14, 23, 59, 0, 0) + + def test_get_previous_day(self): + """Test getting previous day.""" + dt = datetime(2024, 1, 15, 10, 30, 45) + prev_dt = get_previous_day(dt) + assert prev_dt == datetime(2024, 1, 14, 23, 59, 0, 0) + + def test_get_previous_day_month_rollover(self): + """Test getting previous day with month rollover.""" + dt = datetime(2024, 2, 1, 10, 30, 0) + prev_dt = get_previous_day(dt) + assert prev_dt == datetime(2024, 1, 31, 23, 59, 0, 0) + + def test_get_previous_day_year_rollover(self): + """Test getting previous day with year rollover.""" + dt = datetime(2024, 1, 1, 10, 30, 0) + prev_dt = get_previous_day(dt) + assert prev_dt == datetime(2023, 12, 31, 23, 59, 0, 0) + + def test_get_previous_month(self): + """Test getting previous month.""" + dt = datetime(2024, 3, 15, 10, 30, 45) + prev_dt = get_previous_month(dt) + assert prev_dt == datetime(2024, 2, 29, 23, 59, 0, 0) # 2024 is leap year + + def test_get_previous_month_year_rollover(self): + """Test getting previous month with year rollover.""" + dt = datetime(2024, 1, 15, 10, 30, 0) + prev_dt = get_previous_month(dt) + assert prev_dt == datetime(2023, 12, 31, 23, 59, 0, 0) + + def test_get_previous_month_different_days(self): + """Test getting previous month when months have different day counts.""" + # From March to February (non-leap year) + dt = datetime(2023, 3, 31, 10, 30, 0) + prev_dt = get_previous_month(dt) + assert prev_dt == datetime(2023, 2, 28, 23, 59, 0, 0) + + def test_get_previous_month_leap_year(self): + """Test getting previous month in leap year.""" + # From March to February (leap year) + dt = datetime(2024, 3, 31, 10, 30, 0) + prev_dt = get_previous_month(dt) + assert prev_dt == datetime(2024, 2, 29, 23, 59, 0, 0) + + def test_get_previous_year(self): + """Test getting previous year.""" + dt = datetime(2024, 6, 15, 10, 30, 45) + prev_dt = get_previous_year(dt) + assert prev_dt == datetime(2023, 12, 31, 23, 59, 0, 0) + + def test_get_previous_year_from_january(self): + """Test getting previous year from January.""" + dt = datetime(2024, 1, 1, 0, 0, 0) + prev_dt = get_previous_year(dt) + assert prev_dt == datetime(2023, 12, 31, 23, 59, 0, 0) + + +class TestRoundToPreviousMinute: + """Tests for round_to_previous_minute function.""" + + def test_round_to_previous_minute_no_change(self): + """Test rounding when already on minute boundary.""" + dt = datetime(2024, 1, 15, 10, 30, 0, 0) + rounded = round_to_previous_minute(dt) + assert rounded == dt + + def test_round_to_previous_minute_with_seconds(self): + """Test rounding with seconds.""" + dt = datetime(2024, 1, 15, 10, 30, 45, 0) + rounded = round_to_previous_minute(dt) + assert rounded == datetime(2024, 1, 15, 10, 30, 0, 0) + + def test_round_to_previous_minute_with_microseconds(self): + """Test rounding with microseconds.""" + dt = datetime(2024, 1, 15, 10, 30, 0, 123456) + rounded = round_to_previous_minute(dt) + assert rounded == datetime(2024, 1, 15, 10, 30, 0, 0) + + def test_round_to_previous_minute_with_both(self): + """Test rounding with both seconds and microseconds.""" + dt = datetime(2024, 1, 15, 10, 30, 45, 123456) + rounded = round_to_previous_minute(dt) + assert rounded == datetime(2024, 1, 15, 10, 30, 0, 0) + + +class TestBoundaryConditions: + """Tests for boundary conditions in previous time functions.""" + + def test_previous_minute_at_year_start(self): + """Test previous minute at start of year.""" + dt = datetime(2024, 1, 1, 0, 0, 0) + prev_dt = get_previous_minute(dt) + assert prev_dt == datetime(2023, 12, 31, 23, 59, 0) + + def test_previous_hour_at_year_start(self): + """Test previous hour at start of year.""" + dt = datetime(2024, 1, 1, 0, 30, 0) + prev_dt = get_previous_hour(dt) + assert prev_dt == datetime(2023, 12, 31, 23, 59, 0, 0) + + def test_previous_day_at_year_start(self): + """Test previous day at start of year.""" + dt = datetime(2024, 1, 1, 12, 0, 0) + prev_dt = get_previous_day(dt) + assert prev_dt == datetime(2023, 12, 31, 23, 59, 0, 0) + + def test_previous_month_february_to_january_31(self): + """Test previous month from February goes to January 31.""" + dt = datetime(2024, 2, 15, 10, 0, 0) + prev_dt = get_previous_month(dt) + assert prev_dt == datetime(2024, 1, 31, 23, 59, 0, 0) + + def test_previous_month_may_to_april(self): + """Test previous month from May (31 days) to April (30 days).""" + dt = datetime(2024, 5, 31, 10, 0, 0) + prev_dt = get_previous_month(dt) + assert prev_dt == datetime(2024, 4, 30, 23, 59, 0, 0) + + def test_previous_month_july_to_june(self): + """Test previous month from July (31 days) to June (30 days).""" + dt = datetime(2024, 7, 31, 10, 0, 0) + prev_dt = get_previous_month(dt) + assert prev_dt == datetime(2024, 6, 30, 23, 59, 0, 0) + + def test_previous_functions_preserve_time_components(self): + """Test that previous functions preserve appropriate time components.""" + # Previous minute preserves seconds + dt = datetime(2024, 1, 15, 10, 30, 45, 123456) + prev_minute = get_previous_minute(dt) + assert prev_minute.second == 45 + assert prev_minute.microsecond == 123456 + + # Previous hour sets minute to 59, second to 0 + prev_hour = get_previous_hour(dt) + assert prev_hour.minute == 59 + assert prev_hour.second == 0 + assert prev_hour.microsecond == 0 + + # Previous day sets time to 23:59:00 + prev_day = get_previous_day(dt) + assert prev_day.hour == 23 + assert prev_day.minute == 59 + assert prev_day.second == 0 + assert prev_day.microsecond == 0 \ No newline at end of file