|
| 1 | +# (C) 2026 GoodData Corporation |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +from types import SimpleNamespace |
| 5 | +from unittest.mock import MagicMock |
| 6 | + |
| 7 | +from gooddata_sdk.catalog.identifier import CatalogAssigneeIdentifier |
| 8 | +from gooddata_sdk.catalog.organization.entity_model.ip_allowlist_policy import ( |
| 9 | + CatalogIpAllowlistPolicy, |
| 10 | + CatalogIpAllowlistPolicyTargets, |
| 11 | +) |
| 12 | +from gooddata_sdk.catalog.organization.service import CatalogOrganizationService |
| 13 | + |
| 14 | + |
| 15 | +def _make_service() -> tuple[CatalogOrganizationService, MagicMock, MagicMock]: |
| 16 | + """Build a service whose entities-api and actions-api sides are fully mocked.""" |
| 17 | + fake_entities_api = MagicMock(name="EntitiesApi") |
| 18 | + fake_actions_api = MagicMock(name="ActionsApi") |
| 19 | + fake_client = SimpleNamespace( |
| 20 | + entities_api=fake_entities_api, |
| 21 | + layout_api=MagicMock(name="LayoutApi"), |
| 22 | + actions_api=fake_actions_api, |
| 23 | + user_management_api=MagicMock(name="UserManagementApi"), |
| 24 | + ) |
| 25 | + service = CatalogOrganizationService(fake_client) # type: ignore[arg-type] |
| 26 | + return service, fake_entities_api, fake_actions_api |
| 27 | + |
| 28 | + |
| 29 | +def test_ip_allowlist_policy_from_api_reads_attributes_and_relationships() -> None: |
| 30 | + policy = CatalogIpAllowlistPolicy.from_api( |
| 31 | + { |
| 32 | + "id": "corp-vpn-only", |
| 33 | + "type": "ipAllowlistPolicy", |
| 34 | + "attributes": {"allowedSources": ["203.0.113.10/32", "198.51.100.0/24"]}, |
| 35 | + "relationships": { |
| 36 | + "users": {"data": [{"id": "admin", "type": "user"}]}, |
| 37 | + "userGroups": {"data": [{"id": "admins", "type": "userGroup"}]}, |
| 38 | + }, |
| 39 | + } |
| 40 | + ) |
| 41 | + |
| 42 | + assert policy.id == "corp-vpn-only" |
| 43 | + assert policy.allowed_sources == ["203.0.113.10/32", "198.51.100.0/24"] |
| 44 | + assert policy.users == [CatalogAssigneeIdentifier(id="admin", type="user")] |
| 45 | + assert policy.user_groups == [CatalogAssigneeIdentifier(id="admins", type="userGroup")] |
| 46 | + |
| 47 | + |
| 48 | +def test_ip_allowlist_policy_from_api_defaults_missing_optional_fields() -> None: |
| 49 | + policy = CatalogIpAllowlistPolicy.from_api({"id": "corp-vpn-only", "type": "ipAllowlistPolicy"}) |
| 50 | + |
| 51 | + assert policy.id == "corp-vpn-only" |
| 52 | + assert policy.allowed_sources == [] |
| 53 | + assert policy.users == [] |
| 54 | + assert policy.user_groups == [] |
| 55 | + |
| 56 | + |
| 57 | +def test_ip_allowlist_policy_to_api_uses_json_api_shape() -> None: |
| 58 | + policy = CatalogIpAllowlistPolicy( |
| 59 | + id="corp-vpn-only", |
| 60 | + allowed_sources=["203.0.113.10/32"], |
| 61 | + users=[CatalogAssigneeIdentifier(id="admin", type="user")], |
| 62 | + user_groups=[CatalogAssigneeIdentifier(id="admins", type="userGroup")], |
| 63 | + ) |
| 64 | + |
| 65 | + document = policy.to_api() |
| 66 | + data = document.data |
| 67 | + |
| 68 | + assert data.id == "corp-vpn-only" |
| 69 | + assert data.type == "ipAllowlistPolicy" |
| 70 | + assert data.attributes.allowed_sources == ["203.0.113.10/32"] |
| 71 | + users = data.relationships.users.data.value |
| 72 | + assert [(user.id, user.type) for user in users] == [("admin", "user")] |
| 73 | + user_groups = data.relationships.user_groups.data.value |
| 74 | + assert [(group.id, group.type) for group in user_groups] == [("admins", "userGroup")] |
| 75 | + |
| 76 | + |
| 77 | +def test_ip_allowlist_policy_to_api_omits_empty_relationships() -> None: |
| 78 | + policy = CatalogIpAllowlistPolicy(id="corp-vpn-only", allowed_sources=["203.0.113.10/32"]) |
| 79 | + |
| 80 | + document = policy.to_api() |
| 81 | + |
| 82 | + assert document.data.attributes.allowed_sources == ["203.0.113.10/32"] |
| 83 | + assert "relationships" not in document.data |
| 84 | + |
| 85 | + |
| 86 | +def test_ip_allowlist_targets_to_api_uses_action_payload_shape() -> None: |
| 87 | + targets = CatalogIpAllowlistPolicyTargets( |
| 88 | + targets=[ |
| 89 | + CatalogAssigneeIdentifier(id="admin", type="user"), |
| 90 | + CatalogAssigneeIdentifier(id="admins", type="userGroup"), |
| 91 | + ] |
| 92 | + ) |
| 93 | + |
| 94 | + payload = targets.to_api() |
| 95 | + |
| 96 | + assert [(target.id, target.type) for target in payload.targets] == [ |
| 97 | + ("admin", "user"), |
| 98 | + ("admins", "userGroup"), |
| 99 | + ] |
| 100 | + |
| 101 | + |
| 102 | +def test_ip_allowlist_policy_crud_methods_call_generated_client() -> None: |
| 103 | + service, entities_api, _ = _make_service() |
| 104 | + policy_out = SimpleNamespace( |
| 105 | + data={ |
| 106 | + "id": "corp-vpn-only", |
| 107 | + "type": "ipAllowlistPolicy", |
| 108 | + "attributes": {"allowedSources": ["203.0.113.10/32"]}, |
| 109 | + } |
| 110 | + ) |
| 111 | + entities_api.get_entity_ip_allowlist_policies.return_value = policy_out |
| 112 | + entities_api.create_entity_ip_allowlist_policies.return_value = policy_out |
| 113 | + entities_api.update_entity_ip_allowlist_policies.return_value = policy_out |
| 114 | + |
| 115 | + policy = CatalogIpAllowlistPolicy(id="corp-vpn-only", allowed_sources=["203.0.113.10/32"]) |
| 116 | + |
| 117 | + assert service.get_ip_allowlist_policy("corp-vpn-only").id == "corp-vpn-only" |
| 118 | + assert service.create_ip_allowlist_policy(policy).allowed_sources == ["203.0.113.10/32"] |
| 119 | + assert service.update_ip_allowlist_policy(policy).id == "corp-vpn-only" |
| 120 | + service.delete_ip_allowlist_policy("corp-vpn-only") |
| 121 | + |
| 122 | + assert entities_api.get_entity_ip_allowlist_policies.call_args.args[0] == "corp-vpn-only" |
| 123 | + create_doc = entities_api.create_entity_ip_allowlist_policies.call_args.kwargs[ |
| 124 | + "json_api_ip_allowlist_policy_in_document" |
| 125 | + ] |
| 126 | + assert create_doc.data.id == "corp-vpn-only" |
| 127 | + assert entities_api.update_entity_ip_allowlist_policies.call_args.args[0] == "corp-vpn-only" |
| 128 | + assert entities_api.delete_entity_ip_allowlist_policies.call_args.args[0] == "corp-vpn-only" |
| 129 | + |
| 130 | + |
| 131 | +def test_list_ip_allowlist_policies_loads_all_entities() -> None: |
| 132 | + service, entities_api, _ = _make_service() |
| 133 | + entities_api.get_all_entities_ip_allowlist_policies.return_value = SimpleNamespace( |
| 134 | + data=[ |
| 135 | + {"id": "first", "type": "ipAllowlistPolicy"}, |
| 136 | + {"id": "second", "type": "ipAllowlistPolicy"}, |
| 137 | + ], |
| 138 | + included=[], |
| 139 | + links=SimpleNamespace(next=None), |
| 140 | + ) |
| 141 | + |
| 142 | + policies = service.list_ip_allowlist_policies() |
| 143 | + |
| 144 | + assert [policy.id for policy in policies] == ["first", "second"] |
| 145 | + assert entities_api.get_all_entities_ip_allowlist_policies.called |
| 146 | + |
| 147 | + |
| 148 | +def test_ip_allowlist_target_actions_call_generated_client() -> None: |
| 149 | + service, _, actions_api = _make_service() |
| 150 | + targets = CatalogIpAllowlistPolicyTargets(targets=[CatalogAssigneeIdentifier(id="admin", type="user")]) |
| 151 | + |
| 152 | + service.add_targets_to_ip_allowlist_policy("corp-vpn-only", targets) |
| 153 | + service.remove_targets_from_ip_allowlist_policy("corp-vpn-only", targets) |
| 154 | + |
| 155 | + add_targets = actions_api.add_targets.call_args.args[1].targets |
| 156 | + assert actions_api.add_targets.call_args.args[0] == "corp-vpn-only" |
| 157 | + assert [(target.id, target.type) for target in add_targets] == [("admin", "user")] |
| 158 | + remove_targets = actions_api.remove_targets.call_args.args[1].targets |
| 159 | + assert actions_api.remove_targets.call_args.args[0] == "corp-vpn-only" |
| 160 | + assert [(target.id, target.type) for target in remove_targets] == [("admin", "user")] |
0 commit comments