# By: Md. Fahim Bin Amin # Contribution: Asif Al Shahariar """ Routine-related fixtures and data factories for pytest. This module provides reusable test fixtures and factories for constructing sample data dictionaries and mock objects required by the Routine class and its related tests. """ from datetime import timedelta, datetime as dt from faker import Faker from objects.routine import Routine from objects.routine_layer import RoutineLayer from objects.routine_rotation import RoutineRotation from unittest.mock import MagicMock from utils import var_names, constants import datetime import factory import pytest import random import uuid fake = Faker() # ────────────────────────────────────────────────────────────── # Mock Constants # ────────────────────────────────────────────────────────────── @pytest.fixture def faker(): """Provides a shared Faker instance for tests.""" return Faker() @pytest.fixture def mock_var_names(): """Mocked object that replicates the string constants found in var_names.""" class MockVarNames: # Layer-related layer = "layer" layer_name = "layer_name" valid_start = "valid_start" valid_end = "valid_end" is_exception = "is_exception" rotation_start = "rotation_start" shift_length = "shift_length" rotation_period = "rotation_period" rotation_frequency = "rotation_frequency" skip_days = "skip_days" rotations = "rotations" start_timestamp = "start_timestamp" end_timestamp = "end_timestamp" # Routine-related routine_id = "routine_id" organization_id = "organization_id" routine_name = "routine_name" timezone = "timezone" routine_layers = "routine_layers" routine_ref_id = "routine_ref_id" associated_policies = "associated_policies" reference_id = "reference_id" # Rotation-related rotation_end = "rotation_end" rotation_ref_id = "rotation_ref_id" start_period = "start_period" end_period = "end_period" display_name = "display_name" assignee_name = "assignee_name" assignee_policy_id = "assignee_policy_id" # On-call metadata on_call = "on_call" return MockVarNames() # ────────────────────────────────────────────────────────────── # Factory: RoutineLayerDictFactory (dict-based) # ────────────────────────────────────────────────────────────── class RoutineLayerDictFactory(factory.Factory): """ Generates a dictionary that mimics the input format for creating a RoutineLayer. """ class Meta: model = dict as_json = False layer_name = factory.LazyFunction(fake.word) layer = factory.Sequence(lambda n: n) valid_start = factory.LazyFunction(lambda: fake.date_time_this_year()) valid_end = factory.LazyFunction(lambda: fake.date_time_this_year()) is_exception = False rotation_start = "08:00:00" shift_length = "08:00" rotation_period = 7 rotation_frequency = 1 skip_days = None rotations = factory.LazyFunction(list) start_timestamp = factory.LazyFunction(lambda: fake.date_time_this_year()) end_timestamp = factory.LazyFunction(lambda: fake.date_time_this_year()) @factory.post_generation def convert_valid_dates(obj, create, extracted, **kwargs): ts_fmt = constants.json_timestamp_format if hasattr(constants, "json_timestamp_format") else "%Y-%m-%dT%H:%M:%S" for field in ["valid_start", "valid_end", "start_timestamp", "end_timestamp"]: if isinstance(obj[field], datetime.datetime): obj[field] = obj[field].strftime(ts_fmt) # ────────────────────────────────────────────────────────────── # Factory: RoutineLayerFactory (object-based) # ────────────────────────────────────────────────────────────── class RoutineLayerFactory(factory.Factory): """ Factory to create RoutineLayer instances with realistic randomized attributes. """ class Meta: model = RoutineLayer layer = factory.Sequence(lambda n: n) layer_name = factory.LazyFunction(fake.word) valid_start = factory.LazyFunction(lambda: fake.date_time_between(start_date='-30d', end_date='-1d')) valid_end = factory.LazyFunction(lambda: fake.date_time_between(start_date='now', end_date='+30d')) is_exception = factory.LazyFunction(lambda: random.choice([True, False])) rotation_start = factory.LazyFunction(lambda: fake.time_object()) shift_length = factory.LazyFunction(lambda: timedelta(hours=random.randint(1, 12))) rotation_period = factory.LazyFunction(lambda: random.choice([7, 14, 28])) rotation_frequency = factory.LazyFunction(lambda: random.randint(1, 3)) skip_days = factory.LazyFunction(lambda: random.sample(range(7), k=random.randint(0, 3))) @factory.lazy_attribute def rotations(self): return [RoutineRotationFactory() for _ in range(2)] start_timestamp = factory.LazyFunction(lambda: fake.date_time_between(start_date='-30d', end_date='now')) end_timestamp = factory.LazyFunction(lambda: fake.date_time_between(start_date='now', end_date='+365d')) # ────────────────────────────────────────────────────────────── # Factory: RoutineRotationFactory (object-based) # ────────────────────────────────────────────────────────────── class RoutineRotationFactory(factory.Factory): """ Factory to create RoutineRotation instances with random test data. """ class Meta: model = RoutineRotation layer = factory.Faker("random_int", min=0, max=2) start_period = factory.Sequence(lambda n: n * 2 + 1) end_period = factory.LazyAttribute(lambda o: o.start_period + random.randint(1, 3)) assignee_name = factory.Faker("user_name") display_name = factory.Faker("name") assignee_policy_id = factory.Faker("uuid4") # ────────────────────────────────────────────────────────────── # Factory: RoutineFactory (object-based) # ────────────────────────────────────────────────────────────── class RoutineFactory(factory.Factory): """ Factory to generate Routine objects composed of nested RoutineLayer instances. """ class Meta: model = Routine routine_id = factory.LazyFunction(lambda: fake.uuid4()) organization_id = factory.LazyFunction(lambda: fake.uuid4()) routine_name = factory.LazyFunction(lambda: fake.company()) routine_timezone = factory.LazyFunction(lambda: random.choice(['UTC', 'Asia/Dhaka', 'US/Eastern'])) routine_layers = factory.List([factory.SubFactory(RoutineLayerFactory) for _ in range(2)]) reference_id = factory.LazyFunction(lambda: fake.uuid4()) associated_policies = factory.LazyFunction(lambda: [fake.uuid4() for _ in range(random.randint(0, 3))]) # ────────────────────────────────────────────────────────────── # Factory: RoutineDictFactory (dict-based) # ────────────────────────────────────────────────────────────── class RoutineDictFactory(factory.Factory): """ Factory to generate Routine dictionaries (for dict-based tests). """ class Meta: model = dict as_json = False routine_id = factory.LazyFunction(lambda: fake.uuid4()) organization_id = factory.LazyFunction(lambda: fake.uuid4()) routine_name = factory.LazyFunction(lambda: fake.company()) timezone = factory.LazyFunction(lambda: random.choice(['UTC', 'Asia/Dhaka', 'US/Eastern'])) # <-- FIXED routine_layers = factory.LazyFunction(list) reference_id = factory.LazyFunction(lambda: fake.uuid4()) associated_policies = factory.LazyFunction(lambda: [fake.uuid4() for _ in range(random.randint(0, 3))]) # ────────────────────────────────────────────────────────────── # Factory: UserFactory (dict-based) # ────────────────────────────────────────────────────────────── class UserFactory(factory.Factory): """ Factory to generate mock user dictionaries for use in on-call rotations. """ class Meta: model = dict id = factory.LazyFunction(lambda: fake.uuid4()) name = factory.LazyFunction(lambda: fake.name()) # ────────────────────────────────────────────────────────────── # Factory: OnCallEntryFactory (dict-based) # ────────────────────────────────────────────────────────────── class OnCallEntryFactory(factory.Factory): """ Factory to generate dictionary-based OnCallEntry records. """ class Meta: model = dict assignee_policy_id = factory.LazyFunction(lambda: fake.uuid4()) user = factory.SubFactory(UserFactory) start_time = factory.LazyFunction(lambda: fake.date_time_this_month()) end_time = factory.LazyFunction(lambda: fake.date_time_this_month()) # ────────────────────────────────────────────────────────────── # Factory: OnCallEntryTupleFactory (tuple-based) # ────────────────────────────────────────────────────────────── class OnCallEntryTupleFactory(factory.Factory): """ Factory to simulate on-call entry tuples: (start_time, end_time, assignee_id) """ class Meta: model = tuple start_time = factory.LazyFunction(lambda: datetime.datetime.now()) end_time = factory.LazyAttribute(lambda o: o.start_time + datetime.timedelta(hours=8)) assignee_id = factory.LazyFunction(lambda: fake.uuid4()) @classmethod def _create(cls, model_class, *args, **kwargs): return (kwargs["start_time"], kwargs["end_time"], kwargs["assignee_id"]) # ────────────────────────────────────────────────────────────── # Sample Dict Fixtures for Routine/RoutineLayer # ────────────────────────────────────────────────────────────── @pytest.fixture def sample_layer_data(): """Dynamically generates a valid RoutineLayer dictionary using RoutineLayerFactory.""" layer_obj = RoutineLayerFactory() layer_dict = { var_names.layer: layer_obj.layer, var_names.layer_name: layer_obj.layer_name, var_names.valid_start: layer_obj.valid_start.isoformat(), var_names.valid_end: layer_obj.valid_end.isoformat(), var_names.rotation_start: layer_obj.rotation_start.strftime("%H:%M:%S"), var_names.shift_length: str(layer_obj.shift_length), var_names.rotation_period: layer_obj.rotation_period, var_names.rotation_frequency: layer_obj.rotation_frequency, var_names.skip_days: layer_obj.skip_days, var_names.is_exception: layer_obj.is_exception, var_names.rotations: [ { var_names.start_period: rot.start_period, var_names.end_period: rot.end_period, var_names.assignee_name: rot.assignee_name, var_names.display_name: rot.display_name, var_names.assignee_policy_id: rot.assignee_policy_id } for rot in layer_obj.rotations ] } return layer_dict @pytest.fixture def sample_routine_data(mock_var_names): """Provides a fully populated routine dictionary using RoutineDictFactory.""" layer_dict = RoutineLayerDictFactory(as_json=True) return { mock_var_names.routine_id: fake.uuid4(), mock_var_names.organization_id: fake.uuid4(), mock_var_names.routine_name: fake.word(), mock_var_names.timezone: "Asia/Dhaka", mock_var_names.routine_layers: [layer_dict], mock_var_names.routine_ref_id: fake.uuid4(), mock_var_names.reference_id: fake.uuid4(), mock_var_names.associated_policies: [fake.uuid4(), fake.uuid4()], } @pytest.fixture def sample_routine_data_missing_optional(mock_var_names): """Routine dictionary excluding optional fields like reference_id, associated_policies.""" layer_dict = RoutineLayerDictFactory(as_json=True) return { mock_var_names.routine_id: fake.uuid4(), mock_var_names.organization_id: fake.uuid4(), mock_var_names.routine_name: fake.word(), mock_var_names.timezone: "Asia/Dhaka", mock_var_names.routine_layers: [layer_dict], } @pytest.fixture def sample_routine_data_for_display(mock_var_names): """Routine data with routine_ref_id populated to simulate for_display=True input.""" layer_dict = RoutineLayerDictFactory(as_json=True) return { mock_var_names.routine_id: fake.uuid4(), mock_var_names.organization_id: fake.uuid4(), mock_var_names.routine_name: fake.word(), mock_var_names.timezone: "Asia/Dhaka", mock_var_names.routine_layers: [layer_dict], mock_var_names.routine_ref_id: "ref-123", } @pytest.fixture def sample_routine_data_missing_required(mock_var_names): """Deliberately omits required fields (e.g., routine_id, routine_layers).""" return { mock_var_names.organization_id: fake.uuid4(), mock_var_names.routine_name: fake.word(), mock_var_names.timezone: "Asia/Dhaka", } @pytest.fixture def sample_routine_data_malformed_layer(mock_var_names): """Returns a routine dict with invalid routine_layers (non-list type).""" return { mock_var_names.routine_id: fake.uuid4(), mock_var_names.organization_id: fake.uuid4(), mock_var_names.routine_name: fake.word(), mock_var_names.timezone: "Asia/Dhaka", mock_var_names.routine_layers: "not_a_list", mock_var_names.routine_ref_id: fake.uuid4(), } # ────────────────────────────────────────────────────────────── # Additional advanced fixtures and scenario matrices # (gap_adjustment_cases, gaps_between_schedule_scenarios, boundary_gap_cases, etc.) # ────────────────────────────────────────────────────────────── @pytest.fixture def gap_adjustment_scenarios(mock_var_names): """ Test-case matrix for Routine.get_gap_adjusted_schedule. Each dict contains: desc – human-readable description item – original {rotation_start, rotation_end, on_call} gap – (gap_start, gap_end) tuple expected – expected result after adjustment: • None → item is outside gap • dict → adjusted schedule inside the gap """ base = dt(2000, 1, 1) return [ { "desc": "No overlap — item ends before gap starts", "item": { mock_var_names.rotation_start: base + timedelta(hours=0), mock_var_names.rotation_end: base + timedelta(hours=4), mock_var_names.on_call: "user" }, "gap": ( base + timedelta(hours=5), base + timedelta(hours=10) ), "expected": None, }, { "desc": "Full overlap — item fully inside gap", "item": { mock_var_names.rotation_start: base + timedelta(hours=6), mock_var_names.rotation_end: base + timedelta(hours=8), mock_var_names.on_call: "user" }, "gap": ( base + timedelta(hours=5), base + timedelta(hours=10) ), "expected": { mock_var_names.rotation_start: base + timedelta(hours=6), mock_var_names.rotation_end: base + timedelta(hours=8), mock_var_names.on_call: "user", }, }, { "desc": "Left side overlap — gap starts inside item", "item": { mock_var_names.rotation_start: base + timedelta(hours=4), mock_var_names.rotation_end: base + timedelta(hours=7), mock_var_names.on_call: "user" }, "gap": ( base + timedelta(hours=5), base + timedelta(hours=10) ), "expected": { mock_var_names.rotation_start: base + timedelta(hours=5), mock_var_names.rotation_end: base + timedelta(hours=7), mock_var_names.on_call: "user", }, }, { "desc": "Right side overlap — gap ends inside item", "item": { mock_var_names.rotation_start: base + timedelta(hours=5), mock_var_names.rotation_end: base + timedelta(hours=11), mock_var_names.on_call: "user" }, "gap": ( base + timedelta(hours=3), base + timedelta(hours=7) ), "expected": { mock_var_names.rotation_start: base + timedelta(hours=5), mock_var_names.rotation_end: base + timedelta(hours=7), mock_var_names.on_call: "user", }, }, { "desc": "Gap fully inside item — trim to gap", "item": { mock_var_names.rotation_start: base + timedelta(hours=2), mock_var_names.rotation_end: base + timedelta(hours=10), mock_var_names.on_call: "user" }, "gap": ( base + timedelta(hours=4), base + timedelta(hours=8) ), "expected": { mock_var_names.rotation_start: base + timedelta(hours=4), mock_var_names.rotation_end: base + timedelta(hours=8), mock_var_names.on_call: "user", }, }, ] @pytest.fixture def gaps_between_schedule_scenarios(): """ Dynamically generates test cases for Routine.find_gaps_between_schedules(). Purpose: - Validate detection of time gaps between multiple on-call schedules. - Cover edge cases including no input, overlapping intervals, and scattered blocks. Structure of each test case: { "desc": str, # human-readable scenario description "schedules": List[Dict], # list of schedule dicts with start/end "period_start": datetime.datetime, # schedule window start "period_end": datetime.datetime, # schedule window end "expected": List[Tuple[datetime, datetime]] # list of gap intervals } """ def generate_random_schedules(count): """ Generate N random schedule entries with randomized start and end times. Returns: List[Dict]: Schedule items with assignee, rotation_start, rotation_end """ base = datetime.datetime.now().replace(minute=0, second=0, microsecond=0) schedules = [] for _ in range(count): start = base + datetime.timedelta(hours=random.randint(0, 10)) duration = datetime.timedelta(hours=random.randint(1, 3)) end = start + duration schedules.append({ "assignee": fake.first_name(), "rotation_start": start, "rotation_end": end, }) return schedules def compute_expected_gaps(schedules, period_start, period_end): """ Programmatically compute expected gaps from given schedules and time bounds. Merges overlapping intervals and calculates periods with no coverage. """ if not schedules: return [] # Sort schedule intervals by start time intervals = sorted( [(s["rotation_start"], s["rotation_end"]) for s in schedules], key=lambda x: x[0] ) # Merge overlapping or contiguous intervals merged = [] for start, end in intervals: if not merged or start > merged[-1][1]: merged.append([start, end]) else: merged[-1][1] = max(merged[-1][1], end) # Identify gaps between merged intervals gaps = [] if period_start < merged[0][0]: gaps.append((period_start, merged[0][0])) for i in range(1, len(merged)): prev_end = merged[i - 1][1] curr_start = merged[i][0] if prev_end < curr_start: gaps.append((prev_end, curr_start)) if merged[-1][1] < period_end: gaps.append((merged[-1][1], period_end)) return gaps def generate_case(): """ Create one randomized test case with N schedules and expected gaps. """ num_schedules = random.randint(2, 5) schedules = generate_random_schedules(num_schedules) all_times = [t for s in schedules for t in (s["rotation_start"], s["rotation_end"])] period_start = min(all_times) - datetime.timedelta(hours=random.randint(1, 2)) period_end = max(all_times) + datetime.timedelta(hours=random.randint(1, 2)) expected = compute_expected_gaps(schedules, period_start, period_end) return { "desc": f"{num_schedules} randomized schedule(s) between {period_start} and {period_end}", "schedules": schedules, "period_start": period_start, "period_end": period_end, "expected": expected } # Return test matrix with both static and randomized scenarios return [ { "desc": "No schedules (empty input case)", "schedules": [], "period_start": datetime.datetime.now(), "period_end": datetime.datetime.now() + datetime.timedelta(hours=random.randint(6, 12)), "expected": [] }, generate_case(), generate_case(), generate_case() ] @pytest.fixture def boundary_gap_cases(mock_var_names): """ Dynamically generates boundary-edge test cases for Routine.get_gap_adjusted_schedule(). Focus: - Exact match cases - Partial overlaps (left/right) - Gaps within schedule - Schedule inside gap - No overlap Returns: List[Dict]: Test matrix of case dictionaries with: - desc: human-readable scenario - item: original schedule (dict with start/end/on_call) - gap: (gap_start, gap_end) as datetime tuple - expected: expected adjusted result (or None) """ now = datetime.datetime.now().replace(second=0, microsecond=0) user_id = str(uuid.uuid4()) # Define base schedule range base_start = now base_duration = timedelta(minutes=random.randint(180, 360)) # 3–6 hours base_end = base_start + base_duration # Offset helpers offset_small = timedelta(minutes=random.randint(20, 45)) offset_medium = timedelta(minutes=random.randint(60, 120)) offset_large = timedelta(minutes=random.randint(150, 240)) return [ { "desc": "Exact alignment with gap → included fully", "item": { mock_var_names.rotation_start: base_start, mock_var_names.rotation_end: base_end, mock_var_names.on_call: [user_id] }, "gap": (base_start, base_end), "expected": { mock_var_names.rotation_start: base_start, mock_var_names.rotation_end: base_end, mock_var_names.on_call: [user_id] } }, { "desc": "Gap trims from inside → keep trimmed slice", "item": { mock_var_names.rotation_start: base_start, mock_var_names.rotation_end: base_end, mock_var_names.on_call: [user_id] }, "gap": ( base_start + offset_small, base_end - offset_small ), "expected": { mock_var_names.rotation_start: base_start + offset_small, mock_var_names.rotation_end: base_end - offset_small, mock_var_names.on_call: [user_id] } }, { "desc": "Gap before schedule → no overlap → expect None", "item": { mock_var_names.rotation_start: base_start + offset_large, mock_var_names.rotation_end: base_start + offset_large + offset_medium, mock_var_names.on_call: [user_id] }, "gap": ( base_start - offset_medium, base_start - offset_small ), "expected": None }, { "desc": "Schedule fully inside gap → keep full item", "item": { mock_var_names.rotation_start: base_start + offset_small, mock_var_names.rotation_end: base_start + offset_small + offset_medium, mock_var_names.on_call: [user_id] }, "gap": (base_start, base_end), "expected": { mock_var_names.rotation_start: base_start + offset_small, mock_var_names.rotation_end: base_start + offset_small + offset_medium, mock_var_names.on_call: [user_id] } }, { "desc": "Gap overlaps left part → trim end only", "item": { mock_var_names.rotation_start: base_start, mock_var_names.rotation_end: base_end, mock_var_names.on_call: [user_id] }, "gap": ( base_start, base_start + offset_medium ), "expected": { mock_var_names.rotation_start: base_start, mock_var_names.rotation_end: base_start + offset_medium, mock_var_names.on_call: [user_id] } } ] # ────────────────────────────────────────────────────────────── # Fixture: empty_layer_routine for Routine.to_dict() edge case # ────────────────────────────────────────────────────────────── @pytest.fixture def empty_layer_routine(): """ Fixture that returns a Routine instance with no layers defined. Purpose: Validates edge behavior of Routine.to_dict() when the routine has an empty list of routine_layers. Returns: Routine: Instance with empty routine_layers. """ return Routine( routine_id=str(uuid.uuid4()), organization_id=str(uuid.uuid4()), routine_name=fake.company(), routine_timezone=random.choice(["UTC", "Asia/Dhaka", "Europe/London"]), routine_layers=[], # ⛔ No layers provided → critical edge case reference_id=None, associated_policies=None ) # ────────────────────────────────────────────────────────────── # Fixture: routine_layer_get_on_call_cases # ────────────────────────────────────────────────────────────── @pytest.fixture def routine_layer_get_on_call_cases(): """ Provides a matrix of dynamic test cases for RoutineLayer.get_on_call(). Validates logic across: - Valid vs. invalid time ranges (valid_start/valid_end) - Rotation existence and alignment - skip_days exclusion - Ongoing rotations from previous days Each test case includes: - desc (str): Human-readable test description - layer_config (dict): Attributes to construct a RoutineLayer - check_time (datetime): Time used for is_on_call evaluation - expected_found (bool): Whether an assignee is expected on call """ now = datetime.datetime.now().replace(second=0, microsecond=0) weekday = now.weekday() previous_day = now - timedelta(days=1) def rand_shift(hours_min=1, hours_max=4): """Generates a random shift length timedelta.""" return timedelta(hours=random.randint(hours_min, hours_max)) def rand_offset(mins_min=30, mins_max=90): """Generates a random offset timedelta.""" return timedelta(minutes=random.randint(mins_min, mins_max)) return [ { "desc": "✅ Valid time and user rotation active", "layer_config": { "valid_start": now - timedelta(days=random.randint(1, 2)), "valid_end": now + timedelta(days=random.randint(1, 2)), "rotation_start": now.time(), "shift_length": rand_shift(2, 3), "skip_days": [], "rotations": [RoutineRotationFactory(start_period=1, end_period=5)] }, "check_time": now, "expected_found": True }, { "desc": "❌ Time not in validity window (future layer start)", "layer_config": { "valid_start": now + timedelta(days=1), "valid_end": now + timedelta(days=2), "rotation_start": now.time(), "shift_length": rand_shift(), "skip_days": [], "rotations": [] }, "check_time": now, "expected_found": False }, { "desc": "❌ No rotation active at current time", "layer_config": { "valid_start": now - timedelta(days=1), "valid_end": now + timedelta(days=1), "rotation_start": now.time(), "shift_length": rand_shift(1, 2), "skip_days": [], "rotations": [] }, "check_time": now, "expected_found": False }, { "desc": "❌ Current day is in skip_days", "layer_config": { "valid_start": now - timedelta(days=2), "valid_end": now + timedelta(days=2), "rotation_start": now.time(), "shift_length": rand_shift(2, 3), "skip_days": [weekday], # today skipped "rotations": [RoutineRotationFactory(start_period=1, end_period=5)] }, "check_time": now, "expected_found": False }, { "desc": "✅ Valid rotation from previous day still active", "layer_config": { "valid_start": previous_day, "valid_end": now + timedelta(days=1), "rotation_start": (now - rand_offset(120, 150)).time(), # start 2+ hrs ago "shift_length": rand_shift(3, 4), # still within shift "skip_days": [], "rotations": [RoutineRotationFactory(start_period=1, end_period=5)] }, "check_time": now, "expected_found": True }, { "desc": "❌ Too early — layer is valid but not started yet", "layer_config": { "valid_start": now + timedelta(days=1), "valid_end": now + timedelta(days=3), "rotation_start": now.time(), "shift_length": rand_shift(2, 3), "skip_days": [], "rotations": [RoutineRotationFactory(start_period=1, end_period=5)] }, "check_time": now, "expected_found": False } ] # ────────────────────────────────────────────────────────────── # Fixture: routine_layer_schedule_matrix_cases # ────────────────────────────────────────────────────────────── @pytest.fixture def routine_layer_schedule_matrix_cases(): """ Provides randomized, scenario-driven test cases for RoutineLayer.prepare_schedule(). Each test simulates: - Valid and invalid rotation data - Skipped days that bypass scheduling - Edge-case shifts (non-24hr, multi-day) - Infinite loop prevention logic Returns: list[dict]: Each dict contains: - desc: Description of the case - layer: A RoutineLayer instance - expected_non_empty: Whether schedule should contain entries """ base_date = datetime.datetime.now().replace(hour=9, minute=0, second=0, microsecond=0) common_args = { "valid_start": base_date, "valid_end": base_date + timedelta(days=random.randint(3, 7)), "rotation_start": datetime.time(9, 0), "rotation_frequency": 1, "rotation_period": 5, "shift_length": timedelta(hours=8), "layer": 0, "layer_name": fake.word(), } def create_layer(rotations, skip_days=None, shift_len=None, override_end=None): """ Builds a RoutineLayer with custom overrides for dynamic scenarios. """ return RoutineLayer( rotations=rotations, skip_days=skip_days or [], shift_length=shift_len if shift_len else common_args["shift_length"], start_timestamp=base_date, end_timestamp=override_end or common_args["valid_end"], is_exception=False, **{k: v for k, v in common_args.items() if k != "shift_length"} ) def mock_rotation(name): """Creates a fake RoutineRotation-like mock with unique identifiers.""" return MagicMock( assignee_name=name, display_name=name.title(), assignee_policy_id=str(uuid.uuid4()), start_period=random.randint(1, 3), end_period=random.randint(4, 8), ) return [ { "desc": "✅ Valid rotation generates schedule", "layer": create_layer([mock_rotation("a1")]), "expected_non_empty": True }, { "desc": "❌ No rotations present", "layer": create_layer([]), "expected_non_empty": False }, { "desc": "⛔ All days skipped", "layer": create_layer([mock_rotation("a2")], skip_days=list(range(7))), "expected_non_empty": False }, { "desc": "🔁 Infinite loop safeguard", "layer": create_layer([mock_rotation("loop")]), "expected_non_empty": True }, { "desc": "🌙 Cross-day (28h shift span)", "layer": create_layer([mock_rotation("overlap")], shift_len=timedelta(hours=28)), "expected_non_empty": True }, { "desc": "🕗 Irregular 10hr shifts", "layer": create_layer([mock_rotation("shift10")], shift_len=timedelta(hours=10)), "expected_non_empty": True } ] # ────────────────────────────────────────────────────────────── # Test Stub: Routine.is_on_call() # ────────────────────────────────────────────────────────────── class TestRoutineIsOnCall: """ Test suite for Routine.is_on_call(). Validates that the method correctly detects whether a user is currently assigned on-call based on assignee policy ID. Scenarios: - Assignee is in the on-call list → returns True - Assignee is not in the list → returns False Dependencies: - Routine.get_on_call is patched to isolate logic from underlying schedule layers. """ def test_is_on_call_scenarios(self, subtests): """ Parametric sub-test execution for is_on_call() outcomes. Mocks: - get_on_call() is patched to return a controlled list of users instead of relying on actual schedule computation. """ # Setup: simulate a known matching assignee and a non-matching one assignee_pol_id_present = "policy-abc" assignee_pol_id_absent = "policy-xyz" # Assignee metadata assignee_name = factory.Faker("user_name") display_name = factory.Faker("name") assignee_policy_id = factory.Faker("uuid4") # Define scenario matrix test_cases = [ { "scenario": "Assignee found in on-call list", "on_call_users": [ OnCallEntryFactory(assignee_policy_id=assignee_pol_id_present) ], "check_user_id": assignee_pol_id_present, "expected": True, "why": "Assignee is explicitly listed in mocked on-call data" }, { "scenario": "Assignee not found", "on_call_users": [ OnCallEntryFactory(assignee_policy_id=assignee_pol_id_present) ], "check_user_id": assignee_pol_id_absent, "expected": False, "why": "Assignee is not included in on-call list" } ] for case in test_cases: with subtests.test(msg=case["scenario"]): # Construct a barebones Routine instance; layers irrelevant here routine = Routine( routine_id="r2", organization_id="org2", routine_name="Check On-Call", routine_timezone="UTC", routine_layers=[] ) # Patch get_on_call() to return controlled data routine.get_on_call = MagicMock(return_value=case["on_call_users"]) # Run the actual check result = routine.is_on_call(case["check_user_id"]) # Assertion with detailed failure context assert result == case["expected"], ( f"Scenario failed: {case['scenario']} → {case['why']}" )