# By: Asif Al Shahariar from faker import Faker from tests.fixtures.policy_fixtures import PolicyLevelDictFactory from tests.fixtures.routine_fixtures import RoutineFactory, RoutineDictFactory from unittest.mock import patch import pandas as pd import pytest # Patch pandas.read_csv before anything imports Routine fake_df = pd.DataFrame({'value': ['World', 'Bar']}, index=['Hello', 'Foo']) with patch('pandas.read_csv', return_value=fake_df): from objects.policy import Policy from objects.policy_level import PolicyLevel from objects.routine import Routine from utils import var_names fake = Faker() @pytest.fixture def base_policy_details(): """Returns a base details dict for a policy.""" return { var_names.policy_id: fake.random_int(min=1, max=1000), var_names.organization_id: fake.random_int(min=1, max=10000), var_names.policy_name: fake.company(), var_names.policy_type: "USER", var_names.levels: [PolicyLevelDictFactory()], var_names.policy_ref_id: fake.uuid4(), var_names.tags: [fake.word(), fake.word()] } def make_policy_details(base, **overrides): """Helper to override fields in the base policy details.""" details = base.copy() details.update(overrides) return details def test_create_policy(base_policy_details): policy = Policy.create_policy(base_policy_details) assert isinstance(policy, Policy) assert policy.policy_id == base_policy_details[var_names.policy_id] assert policy.organization_id == base_policy_details[var_names.organization_id] assert policy.policy_name == base_policy_details[var_names.policy_name] assert policy.policy_type == base_policy_details[var_names.policy_type] assert len(policy.levels) == len(base_policy_details[var_names.levels]) assert policy.reference_id == base_policy_details[var_names.policy_ref_id] def test_create_policy_multiple_levels(base_policy_details): overrides = { var_names.levels: [PolicyLevelDictFactory(), PolicyLevelDictFactory()] } details = make_policy_details(base_policy_details, **overrides) policy = Policy.create_policy(details) assert isinstance(policy, Policy) assert len(policy.levels) == 2 assert all(isinstance(level, PolicyLevel) for level in policy.levels) def test_create_policy_with_missing_optional_fields(base_policy_details): details = make_policy_details(base_policy_details) details.pop(var_names.policy_ref_id) details.pop(var_names.tags) policy = Policy.create_policy(details) assert isinstance(policy, Policy) assert policy.reference_id is None assert policy.tags is None def test_get_on_call_valid_level(base_policy_details): policy = Policy.create_policy(base_policy_details) level_number = 1 on_call_assignees = policy.get_on_call(level_number) assert isinstance(on_call_assignees, list) assert on_call_assignees == policy.levels[0].routines[0].get_on_call() def test_get_next_available_level_on_call(base_policy_details): overrides = { var_names.levels: [PolicyLevelDictFactory(), PolicyLevelDictFactory()] } details = make_policy_details(base_policy_details, **overrides) policy = Policy.create_policy(details) level_number = 1 next_level = policy.get_next_available_level_on_call(level_number) assert isinstance(next_level, tuple) assert len(next_level) == 2 assert isinstance(next_level[0], int) assert isinstance(next_level[1], list) def test_get_next_available_level_on_call_no_more_level_available(base_policy_details): policy = Policy.create_policy(base_policy_details) level_number = 2 next_level = policy.get_next_available_level_on_call(level_number) assert isinstance(next_level, tuple) assert next_level[0] == level_number assert isinstance(next_level[1], list) assert len(next_level[1]) == 0 def test_is_user_policy(base_policy_details): policy = Policy.create_policy(base_policy_details) assert policy.is_user_policy() is True def test_is_group_policy(base_policy_details): details = make_policy_details(base_policy_details, **{var_names.policy_type: "GROUP"}) policy = Policy.create_policy(details) assert policy.is_group_policy() is True def test_get_level_minutes(base_policy_details): policy = Policy.create_policy(base_policy_details) level_number = 1 minutes = policy.get_level_minutes(level_number) assert isinstance(minutes, int) assert minutes == policy.levels[0].minutes def test_has_level_with_users_context(base_policy_details): policy = Policy.create_policy(base_policy_details) level_number = 1 has_level = policy.has_level(level_number) assert isinstance(has_level, bool) if not policy.levels[0].routines[0].get_on_call(): assert has_level is False else: assert has_level is True def test_to_dict_basic_info(base_policy_details): policy = Policy.create_policy(base_policy_details) policy_dict = policy.to_dict(basic_info=True) assert isinstance(policy_dict, dict) assert policy_dict[var_names.policy_id] == base_policy_details[var_names.policy_id] assert policy_dict[var_names.policy_name] == base_policy_details[var_names.policy_name] assert policy_dict[var_names.policy_type] == base_policy_details[var_names.policy_type] assert len(policy_dict[var_names.levels]) == len(base_policy_details[var_names.levels]) def test_equivalent_to(base_policy_details): policy1 = Policy.create_policy(base_policy_details) policy2 = Policy.create_policy(base_policy_details) assert policy1.equivalent_to(policy2) is True policy2.policy_name = "Modified Name" assert policy1.equivalent_to(policy2) is False def test_get_routine_valid_id(base_policy_details): from tests.fixtures.routine_fixtures import RoutineDictFactory routine_data = RoutineDictFactory() routine_obj = Routine( routine_id=routine_data.get("routine_id"), organization_id=routine_data.get("organization_id"), routine_name=routine_data.get("routine_name"), routine_timezone=routine_data.get("timezone"), routine_layers=routine_data.get("routine_layers"), reference_id=routine_data.get("reference_id"), associated_policies=routine_data.get("associated_policies"), ) policy = Policy.create_policy(base_policy_details) policy.levels[0].routines.append(routine_obj) routine_id = routine_obj.routine_id routine = policy.get_routine(routine_id) assert isinstance(routine, Routine) assert routine.routine_id == routine_id assert routine_obj in policy.levels[0].routines def test_create_policy_missing_policy_id(base_policy_details): details = make_policy_details(base_policy_details) details.pop(var_names.policy_id) with pytest.raises(KeyError, match="policy_id"): Policy.create_policy(details) def test_create_policy_missing_levels(base_policy_details): details = make_policy_details(base_policy_details) details.pop(var_names.levels) with pytest.raises(KeyError, match="levels"): Policy.create_policy(details) def test_get_on_call_negative_level(base_policy_details): policy = Policy.create_policy(base_policy_details) assert policy.get_on_call(-1) == [] def test_create_policy_levels_not_list(base_policy_details): details = make_policy_details(base_policy_details, **{var_names.levels: PolicyLevelDictFactory()}) with pytest.raises(TypeError): Policy.create_policy(details) def test_get_routine_nonexistent_id(base_policy_details): policy = Policy.create_policy(base_policy_details) routine_id = fake.uuid4() routine = policy.get_routine(routine_id) assert routine is None def test_is_user_policy_unknown_type(base_policy_details): details = make_policy_details(base_policy_details, **{var_names.policy_type: "UNKNOWN"}) policy = Policy.create_policy(details) assert policy.is_user_policy() is False assert policy.is_group_policy() is False