diff --git a/featuremanagement/_defaultfilters.py b/featuremanagement/_defaultfilters.py index cffc0bd..75b8b7a 100644 --- a/featuremanagement/_defaultfilters.py +++ b/featuremanagement/_defaultfilters.py @@ -9,6 +9,7 @@ from email.utils import parsedate_to_datetime from typing import cast, List, Mapping, Optional, Dict, Any from ._featurefilters import FeatureFilter +from ._time_window_filter import Recurrence, is_match, TimeWindowFilterSettings FEATURE_FLAG_NAME_KEY = "feature_name" ROLLOUT_PERCENTAGE_KEY = "RolloutPercentage" @@ -18,6 +19,15 @@ # Time Window Constants START_KEY = "Start" END_KEY = "End" +TIME_WINDOW_FILTER_SETTING_RECURRENCE = "Recurrence" + +# Time Window Exceptions +TIME_WINDOW_FILTER_INVALID = ( + "{}: The {} feature filter is not valid for feature {}. It must specify either {}, {}, or both." +) +TIME_WINDOW_FILTER_INVALID_RECURRENCE = ( + "{}: The {} feature filter is not valid for feature {}. It must specify both {} and {} when Recurrence is not None." +) # Targeting kwargs TARGETED_USER_KEY = "user" @@ -31,6 +41,8 @@ FEATURE_FILTER_NAME_KEY = "Name" IGNORE_CASE_KEY = "ignore_case" +logger = logging.getLogger(__name__) + class TargetingException(Exception): """ @@ -52,19 +64,35 @@ def evaluate(self, context: Mapping[Any, Any], **kwargs: Any) -> bool: :return: True if the current time is within the time window. :rtype: bool """ - start = context.get(PARAMETERS_KEY, {}).get(START_KEY) - end = context.get(PARAMETERS_KEY, {}).get(END_KEY) + start = context.get(PARAMETERS_KEY, {}).get(START_KEY, None) + end = context.get(PARAMETERS_KEY, {}).get(END_KEY, None) + recurrence_data = context.get(PARAMETERS_KEY, {}).get(TIME_WINDOW_FILTER_SETTING_RECURRENCE, None) + recurrence = None current_time = datetime.now(timezone.utc) if not start and not end: - logging.warning("%s: At least one of Start or End is required.", TimeWindowFilter.__name__) + logger.warning( + TIME_WINDOW_FILTER_INVALID, + TimeWindowFilter.__name__, + context.get(FEATURE_FLAG_NAME_KEY), + START_KEY, + END_KEY, + ) return False - start_time = parsedate_to_datetime(start) if start else None - end_time = parsedate_to_datetime(end) if end else None + start_time: Optional[datetime] = parsedate_to_datetime(start) if start else None + end_time: Optional[datetime] = parsedate_to_datetime(end) if end else None + + if (start_time is None or start_time <= current_time) and (end_time is None or current_time < end_time): + return True + + if recurrence_data: + recurrence = Recurrence(recurrence_data) + settings = TimeWindowFilterSettings(start_time, end_time, recurrence) + return is_match(settings, current_time) - return (start_time is None or start_time <= current_time) and (end_time is None or current_time < end_time) + return False @FeatureFilter.alias("Microsoft.Targeting") diff --git a/featuremanagement/_time_window_filter/__init__.py b/featuremanagement/_time_window_filter/__init__.py new file mode 100644 index 0000000..34290bd --- /dev/null +++ b/featuremanagement/_time_window_filter/__init__.py @@ -0,0 +1,9 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +from ._recurrence_evaluator import is_match +from ._models import Recurrence, TimeWindowFilterSettings + +__all__ = ["is_match", "Recurrence", "TimeWindowFilterSettings"] diff --git a/featuremanagement/_time_window_filter/_models.py b/featuremanagement/_time_window_filter/_models.py new file mode 100644 index 0000000..ff3574b --- /dev/null +++ b/featuremanagement/_time_window_filter/_models.py @@ -0,0 +1,149 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +from enum import Enum +from typing import Dict, Any, Optional, List +from datetime import datetime +from dataclasses import dataclass +from email.utils import parsedate_to_datetime + + +class RecurrencePatternType(str, Enum): + """ + The recurrence pattern type. + """ + + DAILY = "Daily" + WEEKLY = "Weekly" + + @staticmethod + def from_str(value: str) -> "RecurrencePatternType": + """ + Get the RecurrencePatternType from the string value. + + :param value: The string value. + :type value: str + :return: The RecurrencePatternType. + :rtype: RecurrencePatternType + """ + if value == "Daily": + return RecurrencePatternType.DAILY + if value == "Weekly": + return RecurrencePatternType.WEEKLY + raise ValueError(f"Invalid value: {value}") + + +class RecurrenceRangeType(str, Enum): + """ + The recurrence range type. + """ + + NO_END = "NoEnd" + END_DATE = "EndDate" + NUMBERED = "Numbered" + + @staticmethod + def from_str(value: str) -> "RecurrenceRangeType": + """ + Get the RecurrenceRangeType from the string value. + + :param value: The string value. + :type value: str + :return: The RecurrenceRangeType. + :rtype: RecurrenceRangeType + """ + if value == "NoEnd": + return RecurrenceRangeType.NO_END + if value == "EndDate": + return RecurrenceRangeType.END_DATE + if value == "Numbered": + return RecurrenceRangeType.NUMBERED + raise ValueError(f"Invalid value: {value}") + + +class RecurrencePattern: # pylint: disable=too-few-public-methods + """ + The recurrence pattern settings. + """ + + days: List[str] = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] + + def __init__(self, pattern_data: Dict[str, Any]): + self.type = RecurrencePatternType.from_str(pattern_data.get("Type", "Daily")) + self.interval = pattern_data.get("Interval", 1) + if self.interval <= 0: + raise ValueError("The interval must be greater than 0.") + # Days of the week are represented as a list of strings of their names. + days_of_week_str = pattern_data.get("DaysOfWeek", []) + + # Days of the week are represented as a list of integers from 0 to 6. + self.days_of_week: List[int] = [] + for day in days_of_week_str: + if day not in self.days: + raise ValueError(f"Invalid value for DaysOfWeek: {day}") + if self.days.index(day) in self.days_of_week: + raise ValueError(f"Duplicate day of the week found: {day}") + self.days_of_week.append(self.days.index(day)) + if pattern_data.get("FirstDayOfWeek") and pattern_data.get("FirstDayOfWeek") not in self.days: + raise ValueError(f"Invalid value for FirstDayOfWeek: {pattern_data.get('FirstDayOfWeek')}") + self.first_day_of_week = self.days.index(pattern_data.get("FirstDayOfWeek", "Sunday")) + + +class RecurrenceRange: # pylint: disable=too-few-public-methods + """ + The recurrence range settings. + """ + + type: RecurrenceRangeType + end_date: Optional[datetime] = None + + def __init__(self, range_data: Dict[str, Any]): + self.type = RecurrenceRangeType.from_str(range_data.get("Type", "NoEnd")) + if range_data.get("EndDate") and isinstance(range_data.get("EndDate"), str): + end_date_str = range_data.get("EndDate", "") + try: + self.end_date = parsedate_to_datetime(end_date_str) if end_date_str else None + except ValueError as e: + raise ValueError(f"Invalid value for EndDate: {end_date_str}") from e + except TypeError as e: + # Python 3.9 and earlier throw TypeError if the string is not in RFC 2822 format. + raise ValueError(f"Invalid value for EndDate: {end_date_str}") from e + self.num_of_occurrences = range_data.get("NumberOfOccurrences", 2**63 - 1) + if self.num_of_occurrences <= 0: + raise ValueError("The number of occurrences must be greater than 0.") + + +class Recurrence: # pylint: disable=too-few-public-methods + """ + The recurrence settings. + """ + + pattern: RecurrencePattern + range: RecurrenceRange + + def __init__(self, recurrence_data: Dict[str, Any]): + self.pattern = RecurrencePattern(recurrence_data.get("Pattern", {})) + self.range = RecurrenceRange(recurrence_data.get("Range", {})) + + +@dataclass +class TimeWindowFilterSettings: + """ + The settings for the time window filter. + """ + + start: Optional[datetime] + end: Optional[datetime] + recurrence: Optional[Recurrence] + + +@dataclass +class OccurrenceInfo: + """ + The information of the previous occurrence. + """ + + previous_occurrence: datetime + num_of_occurrences: int diff --git a/featuremanagement/_time_window_filter/_recurrence_evaluator.py b/featuremanagement/_time_window_filter/_recurrence_evaluator.py new file mode 100644 index 0000000..9034b89 --- /dev/null +++ b/featuremanagement/_time_window_filter/_recurrence_evaluator.py @@ -0,0 +1,125 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +from datetime import datetime, timedelta +from typing import Optional +from ._models import RecurrencePatternType, RecurrenceRangeType, TimeWindowFilterSettings, OccurrenceInfo, Recurrence +from ._recurrence_validator import validate_settings, _get_passed_week_days, _sort_days_of_week + +DAYS_PER_WEEK = 7 +REQUIRED_PARAMETER = "Required parameter: %s" + + +def is_match(settings: TimeWindowFilterSettings, now: datetime) -> bool: + """ + Check if the current time is within the time window filter settings. + + :param TimeWindowFilterSettings settings: The settings for the time window filter. + :param datetime now: The current time. + :return: True if the current time is within the time window filter settings, otherwise False. + :rtype: bool + """ + recurrence = settings.recurrence + if recurrence is None: + raise ValueError(REQUIRED_PARAMETER % "Recurrence") + + start = settings.start + end = settings.end + if start is None or end is None: + raise ValueError(REQUIRED_PARAMETER % "Start or End") + + validate_settings(recurrence, start, end) + + previous_occurrence = _get_previous_occurrence(recurrence, start, now) + if previous_occurrence is None: + return False + + occurrence_end_date = previous_occurrence + (end - start) + return now < occurrence_end_date + + +def _get_previous_occurrence(recurrence: Recurrence, start: datetime, now: datetime) -> Optional[datetime]: + if now < start: + return None + + pattern_type = recurrence.pattern.type + if pattern_type == RecurrencePatternType.DAILY: + occurrence_info = _get_daily_previous_occurrence(recurrence, start, now) + elif pattern_type == RecurrencePatternType.WEEKLY: + occurrence_info = _get_weekly_previous_occurrence(recurrence, start, now) + else: + raise ValueError(f"Unsupported recurrence pattern type: {pattern_type}") + + recurrence_range = recurrence.range + range_type = recurrence_range.type + previous_occurrence = occurrence_info.previous_occurrence + end_date = recurrence_range.end_date + if ( + range_type == RecurrenceRangeType.END_DATE + and previous_occurrence is not None + and end_date is not None + and previous_occurrence > end_date + ): + return None + if ( + range_type == RecurrenceRangeType.NUMBERED + and recurrence_range.num_of_occurrences is not None + and occurrence_info.num_of_occurrences > recurrence_range.num_of_occurrences + ): + return None + + return occurrence_info.previous_occurrence + + +def _get_daily_previous_occurrence(recurrence: Recurrence, start: datetime, now: datetime) -> OccurrenceInfo: + interval = recurrence.pattern.interval + num_of_occurrences = (now - start).days // interval + previous_occurrence = start + timedelta(days=num_of_occurrences * interval) + return OccurrenceInfo(previous_occurrence, num_of_occurrences + 1) + + +def _get_weekly_previous_occurrence(recurrence: Recurrence, start: datetime, now: datetime) -> OccurrenceInfo: + pattern = recurrence.pattern + interval = pattern.interval + first_day_of_first_week = start - timedelta(days=_get_passed_week_days(start.weekday(), pattern.first_day_of_week)) + + number_of_interval = (now - first_day_of_first_week).days // (interval * DAYS_PER_WEEK) + first_day_of_most_recent_occurring_week = first_day_of_first_week + timedelta( + days=number_of_interval * (interval * DAYS_PER_WEEK) + ) + sorted_days_of_week = _sort_days_of_week(pattern.days_of_week, pattern.first_day_of_week) + max_day_offset = _get_passed_week_days(sorted_days_of_week[-1], pattern.first_day_of_week) + min_day_offset = _get_passed_week_days(sorted_days_of_week[0], pattern.first_day_of_week) + num_of_occurrences = number_of_interval * len(sorted_days_of_week) - sorted_days_of_week.index(start.weekday()) + + if now > first_day_of_most_recent_occurring_week + timedelta(days=DAYS_PER_WEEK): + num_of_occurrences += len(sorted_days_of_week) + most_recent_occurrence = first_day_of_most_recent_occurring_week + timedelta(days=max_day_offset) + return OccurrenceInfo(most_recent_occurrence, num_of_occurrences) + + day_with_min_offset = first_day_of_most_recent_occurring_week + timedelta(days=min_day_offset) + if start > day_with_min_offset: + num_of_occurrences = 0 + day_with_min_offset = start + if now < day_with_min_offset: + most_recent_occurrence = ( + first_day_of_most_recent_occurring_week + - timedelta(days=interval * DAYS_PER_WEEK) + + timedelta(days=max_day_offset) + ) + else: + most_recent_occurrence = day_with_min_offset + num_of_occurrences += 1 + + for day in sorted_days_of_week[sorted_days_of_week.index(day_with_min_offset.weekday()) + 1 :]: + day_with_min_offset = first_day_of_most_recent_occurring_week + timedelta( + days=_get_passed_week_days(day, pattern.first_day_of_week) + ) + if now < day_with_min_offset: + break + most_recent_occurrence = day_with_min_offset + num_of_occurrences += 1 + + return OccurrenceInfo(most_recent_occurrence, num_of_occurrences) diff --git a/featuremanagement/_time_window_filter/_recurrence_validator.py b/featuremanagement/_time_window_filter/_recurrence_validator.py new file mode 100644 index 0000000..80eae2d --- /dev/null +++ b/featuremanagement/_time_window_filter/_recurrence_validator.py @@ -0,0 +1,164 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +from datetime import datetime, timedelta +from typing import List +from ._models import RecurrencePatternType, RecurrenceRangeType, Recurrence, RecurrencePattern, RecurrenceRange + + +DAYS_PER_WEEK = 7 +TEN_YEARS = 3650 +RECURRENCE_PATTERN = "Pattern" +RECURRENCE_PATTERN_DAYS_OF_WEEK = "DaysOfWeek" +RECURRENCE_RANGE = "Range" +REQUIRED_PARAMETER = "Required parameter: %s" +OUT_OF_RANGE = "Out of range: %s" +TIME_WINDOW_DURATION_TEN_YEARS = "Time window duration exceeds ten years: %s" +NOT_MATCHED = "Start day does not match any day of the week: %s" +TIME_WINDOW_DURATION_OUT_OF_RANGE = "Time window duration is out of range: %s" + + +def validate_settings(recurrence: Recurrence, start: datetime, end: datetime) -> None: + """ + Validate the settings for the time window filter. + + :param TimeWindowFilterSettings settings: The settings for the time window filter. + :raises ValueError: If the settings are invalid. + """ + if not recurrence: + raise ValueError("Recurrence is required") + + _validate_start_end_parameter(start, end) + _validate_recurrence_pattern(recurrence.pattern, start, end) + _validate_recurrence_range(recurrence.range, start) + + +def _validate_start_end_parameter(start: datetime, end: datetime) -> None: + param_name = "end" + if end > start + timedelta(days=TEN_YEARS): + raise ValueError(TIME_WINDOW_DURATION_TEN_YEARS % param_name) + + +def _validate_recurrence_pattern(pattern: RecurrencePattern, start: datetime, end: datetime) -> None: + pattern_type = pattern.type + + if pattern_type == RecurrencePatternType.DAILY: + _validate_daily_recurrence_pattern(pattern, start, end) + else: + _validate_weekly_recurrence_pattern(pattern, start, end) + + +def _validate_recurrence_range(recurrence_range: RecurrenceRange, start: datetime) -> None: + range_type = recurrence_range.type + if range_type == RecurrenceRangeType.END_DATE: + _validate_end_date(recurrence_range, start) + + +def _validate_daily_recurrence_pattern(pattern: RecurrencePattern, start: datetime, end: datetime) -> None: + # "Start" is always a valid first occurrence for "Daily" pattern. + # Only need to check if time window validated + _validate_time_window_duration(pattern, start, end) + + +def _validate_weekly_recurrence_pattern(pattern: RecurrencePattern, start: datetime, end: datetime) -> None: + _validate_days_of_week(pattern) + + # Check whether "Start" is a valid first occurrence + if start.weekday() not in pattern.days_of_week: + raise ValueError(NOT_MATCHED % start.strftime("%A")) + + # Time window duration must be shorter than how frequently it occurs + _validate_time_window_duration(pattern, start, end) + + # Check whether the time window duration is shorter than the minimum gap between days of week + if not _is_duration_compliant_with_days_of_week(pattern, start, end): + raise ValueError(TIME_WINDOW_DURATION_OUT_OF_RANGE % "Recurrence.Pattern.DaysOfWeek") + + +def _validate_time_window_duration(pattern: RecurrencePattern, start: datetime, end: datetime) -> None: + interval_duration = ( + timedelta(days=pattern.interval) + if pattern.type == RecurrencePatternType.DAILY + else timedelta(days=pattern.interval * DAYS_PER_WEEK) + ) + time_window_duration = end - start + if start > end: + raise ValueError(OUT_OF_RANGE % "The filter start date Start needs to before the End date.") + + if time_window_duration > interval_duration: + raise ValueError(TIME_WINDOW_DURATION_OUT_OF_RANGE % "Recurrence.Pattern.Interval") + + +def _validate_days_of_week(pattern: RecurrencePattern) -> None: + days_of_week = pattern.days_of_week + if not days_of_week: + raise ValueError(REQUIRED_PARAMETER % "Recurrence.Pattern.DaysOfWeek") + + +def _validate_end_date(recurrence_range: RecurrenceRange, start: datetime) -> None: + end_date = recurrence_range.end_date + if end_date and end_date < start: + raise ValueError("The Recurrence.Range.EndDate should be after the Start") + + +def _is_duration_compliant_with_days_of_week(pattern: RecurrencePattern, start: datetime, end: datetime) -> bool: + days_of_week = pattern.days_of_week + if len(days_of_week) == 1: + return True + + # Get the date of first day of the week + today = datetime.now() + first_day_of_week = pattern.first_day_of_week + offset = _get_passed_week_days((today.weekday() + 1) % 7, first_day_of_week) + first_date_of_week = today - timedelta(days=offset) + sorted_days_of_week = _sort_days_of_week(days_of_week, first_day_of_week) + + # Loop the whole week to get the min gap between the two consecutive recurrences + prev_occurrence = first_date_of_week + timedelta( + days=_get_passed_week_days(sorted_days_of_week[0], first_day_of_week) + ) + min_gap = timedelta(days=DAYS_PER_WEEK) + + for day in sorted_days_of_week[1:]: + date = first_date_of_week + timedelta(days=_get_passed_week_days(day, first_day_of_week)) + if prev_occurrence is not None: + current_gap = date - prev_occurrence + min_gap = min(min_gap, current_gap) + prev_occurrence = date + + if pattern.interval == 1: + # It may cross weeks. Check the adjacent week + date = first_date_of_week + timedelta( + days=DAYS_PER_WEEK + _get_passed_week_days(sorted_days_of_week[0], first_day_of_week) + ) + + current_gap = date - prev_occurrence + min_gap = min(min_gap, current_gap) + + time_window_duration = end - start + return min_gap >= time_window_duration + + +def _get_passed_week_days(current_day: int, first_day_of_week: int) -> int: + """ + Get the number of days passed since the first day of the week. + :param int current_day: The current day of the week, where Sunday == 0 ... Saturday == 6. + :param int first_day_of_week: The first day of the week (0-6), where Sunday == 0 ... Saturday == 6. + :return: The number of days passed since the first day of the week. + :rtype: int + """ + return (current_day - first_day_of_week + DAYS_PER_WEEK) % DAYS_PER_WEEK + + +def _sort_days_of_week(days_of_week: List[int], first_day_of_week: int) -> List[int]: + sorted_days = sorted(days_of_week) + if first_day_of_week in sorted_days: + return sorted_days[sorted_days.index(first_day_of_week) :] + sorted_days[: sorted_days.index(first_day_of_week)] + next_closest_day = first_day_of_week + for i in range(7): + if (first_day_of_week + i) % 7 in sorted_days: + next_closest_day = (first_day_of_week + i) % 7 + break + return sorted_days[sorted_days.index(next_closest_day) :] + sorted_days[: sorted_days.index(next_closest_day)] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/time_window_filter/test_recurrence_evaluator.py b/tests/time_window_filter/test_recurrence_evaluator.py new file mode 100644 index 0000000..8dabf49 --- /dev/null +++ b/tests/time_window_filter/test_recurrence_evaluator.py @@ -0,0 +1,281 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +from datetime import datetime +import pytest +from featuremanagement._time_window_filter._recurrence_evaluator import is_match +from featuremanagement._time_window_filter._models import TimeWindowFilterSettings, Recurrence + + +def test_is_match_within_time_window(): + start = datetime(2025, 4, 7, 9, 0, 0) + end = datetime(2025, 4, 7, 17, 0, 0) + now = datetime(2025, 4, 8, 10, 0, 0) + + recurrence = Recurrence( + { + "Pattern": {"Type": "Daily", "Interval": 1}, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + assert is_match(settings, now) is True + + +def test_is_match_outside_time_window(): + start = datetime(2025, 4, 7, 9, 0, 0) + end = datetime(2025, 4, 7, 17, 0, 0) + now = datetime(2025, 4, 7, 18, 0, 0) + + recurrence = Recurrence( + { + "Pattern": {"Type": "Daily", "Interval": 1}, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + assert is_match(settings, now) is False + + +def test_is_match_no_previous_occurrence(): + start = datetime(2025, 4, 7, 9, 0, 0) + end = datetime(2025, 4, 7, 17, 0, 0) + now = datetime(2025, 4, 6, 10, 0, 0) # Before the start time + + recurrence = Recurrence( + { + "Pattern": {"Type": "Daily", "Interval": 1}, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + assert is_match(settings, now) is False + + +def test_is_match_no_recurrence(): + start = datetime(2025, 4, 7, 9, 0, 0) + end = datetime(2025, 4, 7, 17, 0, 0) + now = datetime(2025, 4, 7, 10, 0, 0) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=None) + + with pytest.raises(ValueError, match="Required parameter: Recurrence"): + is_match(settings, now) + + +def test_is_match_missing_start(): + end = datetime(2025, 4, 7, 17, 0, 0) + now = datetime(2025, 4, 7, 10, 0, 0) + + recurrence = Recurrence( + { + "Pattern": {"Type": "Daily", "Interval": 1}, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=None, end=end, recurrence=recurrence) + + with pytest.raises(ValueError, match="Required parameter: Start or End"): + is_match(settings, now) + + +def test_is_match_missing_end(): + start = datetime(2025, 4, 7, 9, 0, 0) + now = datetime(2025, 4, 7, 10, 0, 0) + + recurrence = Recurrence( + { + "Pattern": {"Type": "Daily", "Interval": 1}, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=None, recurrence=recurrence) + + with pytest.raises(ValueError, match="Required parameter: Start or End"): + is_match(settings, now) + + +def test_is_match_weekly_recurrence(): + start = datetime(2025, 4, 7, 9, 0, 0) # Monday + end = datetime(2025, 4, 7, 17, 0, 0) # Monday + now = datetime(2025, 4, 14, 10, 0, 0) # Next Monday + + recurrence = Recurrence( + { + "Pattern": {"Type": "Weekly", "Interval": 1, "DaysOfWeek": ["Monday"], "FirstDayOfWeek": "Monday"}, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + assert is_match(settings, now) is True + + +def test_is_match_end_date_has_passed(): + start = datetime(2025, 4, 7, 9, 0, 0) + end = datetime(2025, 4, 7, 17, 0, 0) + now = datetime(2025, 4, 9, 10, 0, 0) # After the end date + + recurrence = Recurrence( + { + "Pattern": {"Type": "Daily", "Interval": 1}, + "Range": {"Type": "EndDate", "EndDate": "Tue, 8 Apr 2025 10:00:00"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + assert is_match(settings, now) is False + + +def test_is_match_numbered_recurrence(): + start = datetime(2025, 4, 7, 9, 0, 0) + end = datetime(2025, 4, 7, 17, 0, 0) + now = datetime(2025, 4, 8, 10, 0, 0) + + recurrence = Recurrence( + { + "Pattern": {"Type": "Daily", "Interval": 1}, + "Range": {"Type": "Numbered", "NumberOfOccurrences": 2}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + assert is_match(settings, now) is True + now = datetime(2025, 4, 15, 10, 0, 0) + assert is_match(settings, now) is False + + +def test_is_match_weekly_recurrence_with_occurrences_single_day(): + start = datetime(2025, 4, 7, 9, 0, 0) # Monday + end = datetime(2025, 4, 7, 17, 0, 0) # Monday + + recurrence = Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 2, + "DaysOfWeek": ["Monday"], + "FirstDayOfWeek": "Monday", + }, + "Range": {"Type": "Numbered", "NumberOfOccurrences": 2}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + # First occurrence should match + assert is_match(settings, datetime(2025, 4, 7, 10, 0, 0)) is True + + # Second week occurrence shouldn't match + assert is_match(settings, datetime(2025, 4, 14, 10, 0, 0)) is False + + # Third week occurrence should match + assert is_match(settings, datetime(2025, 4, 21, 10, 0, 0)) is True + + # Fourth week occurrence shouldn't match + assert is_match(settings, datetime(2025, 4, 28, 10, 0, 0)) is False + + # Fifth week occurrence shouldn't match, passed the range + assert is_match(settings, datetime(2025, 5, 5, 10, 0, 0)) is False + + +def test_is_match_weekly_recurrence_with_occurrences_multi_day(): + start = datetime(2025, 4, 7, 9, 0, 0) # Monday + end = datetime(2025, 4, 7, 17, 0, 0) # Monday + + recurrence = Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 2, + "DaysOfWeek": ["Monday", "Tuesday"], + "FirstDayOfWeek": "Monday", + }, + "Range": {"Type": "Numbered", "NumberOfOccurrences": 4}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + # Before the start time, should not match + assert is_match(settings, datetime(2025, 4, 7, 8, 0, 0)) is False # Monday + + # First occurrence should match + assert is_match(settings, datetime(2025, 4, 7, 10, 0, 0)) is True # Monday + assert is_match(settings, datetime(2025, 4, 8, 10, 0, 0)) is True # Tuesday + + # Second week occurrence shouldn't match + assert is_match(settings, datetime(2025, 4, 14, 10, 0, 0)) is False # Monday + assert is_match(settings, datetime(2025, 4, 15, 10, 0, 0)) is False # Tuesday + + # Third week occurrence should match + assert is_match(settings, datetime(2025, 4, 21, 10, 0, 0)) is True # Monday + assert is_match(settings, datetime(2025, 4, 22, 10, 0, 0)) is True # Tuesday + + # Fourth week occurrence shouldn't match + assert is_match(settings, datetime(2025, 4, 28, 10, 0, 0)) is False # Monday + assert is_match(settings, datetime(2025, 4, 29, 10, 0, 0)) is False # Tuesday + + # Fifth week occurrence shouldn't match + assert is_match(settings, datetime(2025, 5, 5, 10, 0, 0)) is False # Monday + assert is_match(settings, datetime(2025, 5, 6, 10, 0, 0)) is False # Tuesday + + +def test_weekly_recurrence_start_after_min_offset(): + start = datetime(2025, 4, 9, 9, 0, 0) # Monday + end = datetime(2025, 4, 9, 17, 0, 0) # Monday + now = datetime(2025, 4, 12, 10, 0, 0) # Saturday + + recurrence = Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Monday", "Wednesday"], + "FirstDayOfWeek": "Monday", + }, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + # Verify that the main method is_match correctly handles the scenario + assert is_match(settings, now) is False + assert is_match(settings, start) is True + + +def test_weekly_recurrence_now_before_min_offset(): + start = datetime(2025, 4, 9, 9, 0, 0) # Monday + end = datetime(2025, 4, 9, 17, 0, 0) # Monday + now = datetime(2025, 4, 16, 8, 0, 0) # Saturday + + recurrence = Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Wednesday", "Friday"], + "FirstDayOfWeek": "Monday", + }, + "Range": {"Type": "NoEnd"}, + } + ) + + settings = TimeWindowFilterSettings(start=start, end=end, recurrence=recurrence) + + # Verify that the main method is_match correctly handles the scenario + assert is_match(settings, now) is False diff --git a/tests/time_window_filter/test_recurrence_validator.py b/tests/time_window_filter/test_recurrence_validator.py new file mode 100644 index 0000000..3d22f4f --- /dev/null +++ b/tests/time_window_filter/test_recurrence_validator.py @@ -0,0 +1,321 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +from datetime import timedelta, datetime +import pytest +from featuremanagement._time_window_filter._models import Recurrence +from featuremanagement._time_window_filter._recurrence_validator import validate_settings, _sort_days_of_week + +DATE_FORMAT = "%a, %d %b %Y %H:%M:%S" + +START_STRING = "Mon, 31 Mar 2025 00:00:00" +START = datetime.strptime(START_STRING, DATE_FORMAT) +END_STRING = "Mon, 31 Mar 2025 23:59:59" +END = datetime.strptime(END_STRING, DATE_FORMAT) + + +def valid_daily_pattern(): + return { + "Type": "Daily", + "Interval": 1, + "DaysOfWeek": ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"], + "FirstDayOfWeek": "Sunday", + } + + +def valid_no_end_range(): + return { + "Type": "NoEnd", + "EndDate": None, + "NumberOfOccurrences": 10, + } + + +def valid_daily_recurrence(): + return Recurrence( + { + "Pattern": valid_daily_pattern(), + "Range": valid_no_end_range(), + } + ) + + +def valid_daily_end_date_recurrence(): + return Recurrence( + { + "Pattern": valid_daily_pattern(), + "Range": { + "Type": "EndDate", + "EndDate": (START + timedelta(days=10)).strftime(DATE_FORMAT), + "NumberOfOccurrences": 10, + }, + } + ) + + +def test_validate_settings_valid_daily(): + validate_settings(valid_daily_recurrence(), START, END) + + +def test_validate_settings_valid_daily_end_date(): + validate_settings(valid_daily_end_date_recurrence(), START, END) + + +def test_validate_settings_valid_weekly_one_day(): + validate_settings( + Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Monday"], + "FirstDayOfWeek": "Monday", + }, + "Range": valid_no_end_range(), + } + ), + START, + END, + ) + + +def test_validate_settings_valid_weekly(): + validate_settings( + Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"], + "FirstDayOfWeek": "Monday", + }, + "Range": valid_no_end_range(), + } + ), + START, + END, + ) + + +def test_validate_settings_duration_exceeds_ten_years(): + end = START + timedelta(days=3651) + with pytest.raises(ValueError, match="Time window duration exceeds ten years: end"): + validate_settings(valid_daily_recurrence(), START, end) + + +def test_validate_settings_invalid_start_end(): + start = START + timedelta(days=2) + with pytest.raises(ValueError, match="The filter start date Start needs to before the End date."): + validate_settings(valid_daily_recurrence(), start, END) + + +def test_validate_settings_end_date_in_past(): + end = START - timedelta(days=1) + with pytest.raises(ValueError, match="The filter start date Start needs to before the End date."): + validate_settings(valid_daily_recurrence(), START, end) + + +def test_validate_settings_missing_recurrence(): + with pytest.raises(ValueError, match="Recurrence is required"): + validate_settings(None, START, END) + + +def test_validate_settings_invalid_recurrence_pattern(): + with pytest.raises(ValueError, match="Invalid value: InvalidType"): + Recurrence({"Pattern": {"Type": "InvalidType"}, "Range": {"Type": "NoEnd"}}) + + +def test_validate_settings_weekly_recurrence_invalid_start_day(): + with pytest.raises(ValueError, match="Start day does not match any day of the week: Monday"): + validate_settings( + Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Tuesday", "Wednesday", "Thursday", "Friday"], + "FirstDayOfWeek": "Monday", + }, + "Range": valid_no_end_range(), + } + ), + START, + END, + ) + + +def test_validate_settings_period_too_long(): + end = START + timedelta(days=7) + with pytest.raises(ValueError, match="Time window duration is out of range:"): + validate_settings(valid_daily_recurrence(), START, end) + + +def test_validate_settings_no_days_of_week(): + with pytest.raises(ValueError, match="Required parameter: Recurrence.Pattern.DaysOfWeek"): + validate_settings( + Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "FirstDayOfWeek": "Monday", + }, + "Range": valid_no_end_range(), + } + ), + START, + END, + ) + + +def test_validate_settings_end_date_before_start(): + with pytest.raises(ValueError, match="The Recurrence.Range.EndDate should be after the Start"): + validate_settings(valid_daily_end_date_recurrence(), START + timedelta(days=11), END + timedelta(days=11)) + + +def test_is_duration_compliant_with_days_of_week_false(): + pattern = { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Monday", "Wednesday"], + "FirstDayOfWeek": "Monday", + } + + start = datetime(2025, 4, 7, 9, 0, 0) # Monday + end = datetime(2025, 4, 10, 9, 0, 0) # Wednesday (48 hours duration) + with pytest.raises(ValueError, match="Recurrence.Pattern.DaysOfWeek"): + validate_settings(Recurrence({"Pattern": pattern, "Range": valid_no_end_range()}), start, end) + + +def test_sort_days_of_week(): + days_of_week = [0, 3, 5] # Monday, Thursday, Saturday + sorted_days = _sort_days_of_week(days_of_week, 6) + assert sorted_days == [0, 3, 5] + + days_of_week = [5, 0, 3] # Saturday, Monday, Thursday + sorted_days = _sort_days_of_week(days_of_week, 6) + assert sorted_days == [0, 3, 5] + + days_of_week = [0, 1, 2, 3, 4, 5, 6] # All days of the week + sorted_days = _sort_days_of_week(days_of_week, 6) + assert sorted_days == [6, 0, 1, 2, 3, 4, 5] + + days_of_week = [6, 5, 4, 3, 2, 1, 0] # All days of the week in reverse order + sorted_days = _sort_days_of_week(days_of_week, 6) + assert sorted_days == [6, 0, 1, 2, 3, 4, 5] + + days_of_week = [0, 2, 4, 6] # Monday, Wednesday, Friday, Sunday + sorted_days = _sort_days_of_week(days_of_week, 2) + assert sorted_days == [2, 4, 6, 0] + + days_of_week = [1] # Tuesday + sorted_days = _sort_days_of_week(days_of_week, 0) + assert sorted_days == [1] + + +def test_validate_settings_invalid_days_of_week(): + with pytest.raises(ValueError, match="Invalid value for DaysOfWeek: Thor's Day"): + validate_settings( + Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Thor's Day"], + "FirstDayOfWeek": "Monday", + }, + "Range": valid_no_end_range(), + } + ), + START, + END, + ) + with pytest.raises(ValueError, match="Required parameter: Recurrence.Pattern.DaysOfWeek"): + validate_settings( + Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": [], + "FirstDayOfWeek": "Monday", + }, + "Range": valid_no_end_range(), + } + ), + START, + END, + ) + + +def test_validate_settings_invalid_days_of_week_duplicate(): + with pytest.raises(ValueError, match="Duplicate day of the week found: Monday"): + validate_settings( + Recurrence( + { + "Pattern": { + "Type": "Weekly", + "Interval": 1, + "DaysOfWeek": ["Monday", "Monday"], + "FirstDayOfWeek": "Monday", + }, + "Range": valid_no_end_range(), + } + ), + START, + END, + ) + + +def test_validate_settings_invalid_end_date_format(): + with pytest.raises(ValueError, match="Invalid value for EndDate: invalid-date-format"): + validate_settings( + Recurrence( + { + "Pattern": valid_daily_pattern(), + "Range": { + "Type": "EndDate", + "EndDate": "invalid-date-format", + "NumberOfOccurrences": 10, + }, + } + ), + START, + END, + ) + + +def test_validate_settings_boundary_condition_ten_years(): + end = START + timedelta(days=3650) # Exactly 10 years + recurrence = Recurrence( + { + "Pattern": { + "Type": "Daily", + "Interval": 3651, # Interval greater than the time window length + "DaysOfWeek": ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"], + "FirstDayOfWeek": "Sunday", + }, + "Range": valid_no_end_range(), + } + ) + validate_settings(recurrence, START, end) + recurrence = Recurrence( + { + "Pattern": { + "Type": "Daily", + "Interval": 3652, # Interval greater than the time window length + "DaysOfWeek": ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"], + "FirstDayOfWeek": "Sunday", + }, + "Range": valid_no_end_range(), + } + ) + with pytest.raises(ValueError, match="Time window duration exceeds ten years: end"): + validate_settings(recurrence, START, START + timedelta(days=3652)) + + +def test_validate_settings_boundary_condition_interval(): + end = START + timedelta(days=1) # Exactly matches the interval duration for daily recurrence + validate_settings(valid_daily_recurrence(), START, end) diff --git a/tests/time_window_filter/test_time_window_filter_models.py b/tests/time_window_filter/test_time_window_filter_models.py new file mode 100644 index 0000000..651d43a --- /dev/null +++ b/tests/time_window_filter/test_time_window_filter_models.py @@ -0,0 +1,133 @@ +# ------------------------------------------------------------------------ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. See License.txt in the project root for +# license information. +# ------------------------------------------------------------------------- +from featuremanagement._time_window_filter._models import ( + RecurrencePatternType, + RecurrenceRangeType, + RecurrencePattern, + RecurrenceRange, + Recurrence, +) +from datetime import datetime +import math + + +def test_recurrence_pattern_type(): + assert RecurrencePatternType.from_str("Daily") == RecurrencePatternType.DAILY + assert RecurrencePatternType.from_str("Weekly") == RecurrencePatternType.WEEKLY + try: + RecurrencePatternType.from_str("Invalid") + except ValueError as e: + assert str(e) == "Invalid value: Invalid" + + +def test_recurrence_range_type(): + assert RecurrenceRangeType.from_str("NoEnd") == RecurrenceRangeType.NO_END + assert RecurrenceRangeType.from_str("EndDate") == RecurrenceRangeType.END_DATE + assert RecurrenceRangeType.from_str("Numbered") == RecurrenceRangeType.NUMBERED + try: + RecurrenceRangeType.from_str("Invalid") + except ValueError as e: + assert str(e) == "Invalid value: Invalid" + + +def test_recurrence_pattern(): + pattern = RecurrencePattern({"Type": "Daily", "Interval": 1, "DaysOfWeek": ["Monday", "Tuesday"]}) + assert pattern.type == RecurrencePatternType.DAILY + assert pattern.interval == 1 + assert pattern.days_of_week == [0, 1] + assert pattern.first_day_of_week == 6 + + pattern = RecurrencePattern( + {"Type": "Daily", "Interval": 1, "DaysOfWeek": ["Monday", "Tuesday"], "FirstDayOfWeek": "Monday"} + ) + assert pattern.type == RecurrencePatternType.DAILY + assert pattern.interval == 1 + assert pattern.days_of_week == [0, 1] + assert pattern.first_day_of_week == 0 + + try: + pattern = RecurrencePattern( + {"Type": "Daily", "Interval": 1, "DaysOfWeek": ["Monday", "Tuesday"], "FirstDayOfWeek": "Thor's day"} + ) + except ValueError as e: + assert str(e) == "Invalid value for FirstDayOfWeek: Thor's day" + + pattern = RecurrencePattern({"Type": "Weekly", "Interval": 2, "DaysOfWeek": ["Wednesday"]}) + assert pattern.type == RecurrencePatternType.WEEKLY + assert pattern.interval == 2 + assert pattern.days_of_week == [2] + assert pattern.first_day_of_week == 6 + + try: + pattern = RecurrencePattern({"Type": "Daily", "Interval": 0, "DaysOfWeek": ["Monday", "Tuesday"]}) + except ValueError as e: + assert str(e) == "The interval must be greater than 0." + + try: + pattern = RecurrencePattern({"Type": "Daily", "Interval": 1, "DaysOfWeek": ["Monday", "Thor's day"]}) + except ValueError as e: + assert str(e) == "Invalid value for DaysOfWeek: Thor's day" + + +def test_recurrence_range(): + max_occurrences = 2**63 - 1 + + range = RecurrenceRange({"Type": "NoEnd"}) + assert range.type == RecurrenceRangeType.NO_END + assert range.end_date is None + assert range.num_of_occurrences == max_occurrences + + range = RecurrenceRange({"Type": "EndDate", "EndDate": "Mon, 31 Mar 2025 00:00:00"}) + assert range.type == RecurrenceRangeType.END_DATE + assert range.end_date == datetime(2025, 3, 31, 0, 0, 0) + assert range.num_of_occurrences == max_occurrences + + range = RecurrenceRange({"Type": "Numbered", "NumberOfOccurrences": 10}) + assert range.type == RecurrenceRangeType.NUMBERED + assert range.end_date is None + assert range.num_of_occurrences == 10 + + try: + range = RecurrenceRange({"Type": "NoEnd", "NumberOfOccurrences": -1}) + except ValueError as e: + assert str(e) == "The number of occurrences must be greater than 0." + + try: + range = RecurrenceRange({"Type": "NoEnd", "NumberOfOccurrences": 0}) + except ValueError as e: + assert str(e) == "The number of occurrences must be greater than 0." + + try: + range = RecurrenceRange({"Type": "EndDate", "EndDate": "InvalidDate"}) + except ValueError as e: + assert str(e) == "Invalid value for EndDate: InvalidDate" + + +def test_recurrence(): + recurrence = Recurrence( + {"Pattern": {"Type": "Daily", "Interval": 1, "DaysOfWeek": ["Monday"]}, "Range": {"Type": "NoEnd"}} + ) + assert recurrence.pattern.type == RecurrencePatternType.DAILY + assert recurrence.range.type == RecurrenceRangeType.NO_END + + try: + recurrence = Recurrence({"Pattern": {"Type": "Invalid"}, "Range": {"Type": "NoEnd"}}) + except ValueError as e: + assert str(e) == "Invalid value: Invalid" + + try: + recurrence = Recurrence( + {"Pattern": {"Type": "Daily", "Interval": 0, "DaysOfWeek": ["Monday"]}, "Range": {"Type": "NoEnd"}} + ) + except ValueError as e: + assert str(e) == "The interval must be greater than 0." + + try: + recurrence = Recurrence( + {"Pattern": {"Type": "Daily", "Interval": 1, "DaysOfWeek": ["Invalid"]}, "Range": {"Type": "NoEnd"}} + ) + except ValueError as e: + assert str(e) == "Invalid value for DaysOfWeek: Invalid"