From 33a7a69cb8c128b4453c06af9dac8a3060479da5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Harbuz?= Date: Wed, 6 May 2026 10:36:05 +0400 Subject: [PATCH] feat: Implement IAM v1 User Management API - Added AccountUsersService for managing users in an account. - Implemented methods for listing, creating, retrieving, updating, and deleting users. - Created response models: User, UserList, UserCreationResponse, UserGroups, and Group. - Added tests for user management functionalities including listing, creating, updating, and deleting users. - Introduced mock responses for testing HTTP requests in the user management context. --- README.md | 12 +- dynatrace/account/iam_v1/__init__.py | 1 + dynatrace/account/iam_v1/groups.py | 165 +++++++++ dynatrace/account/iam_v1/platform_tokens.py | 145 ++++++++ dynatrace/account/iam_v1/policies.py | 367 ++++++++++++++++++++ dynatrace/account/iam_v1/service_users.py | 127 +++++++ dynatrace/account/iam_v1/users.py | 176 ++++++++++ dynatrace/account/schemas.py | 16 + pyproject.toml | 2 +- test/account/iam_v1/__init__.py | 1 + test/account/iam_v1/test_groups.py | 225 ++++++++++++ test/account/iam_v1/test_platform_tokens.py | 219 ++++++++++++ test/account/iam_v1/test_policies.py | 346 ++++++++++++++++++ test/account/iam_v1/test_service_users.py | 226 ++++++++++++ test/account/iam_v1/test_users.py | 244 +++++++++++++ 15 files changed, 2265 insertions(+), 7 deletions(-) create mode 100644 dynatrace/account/iam_v1/__init__.py create mode 100644 dynatrace/account/iam_v1/groups.py create mode 100644 dynatrace/account/iam_v1/platform_tokens.py create mode 100644 dynatrace/account/iam_v1/policies.py create mode 100644 dynatrace/account/iam_v1/service_users.py create mode 100644 dynatrace/account/iam_v1/users.py create mode 100644 test/account/iam_v1/__init__.py create mode 100644 test/account/iam_v1/test_groups.py create mode 100644 test/account/iam_v1/test_platform_tokens.py create mode 100644 test/account/iam_v1/test_policies.py create mode 100644 test/account/iam_v1/test_service_users.py create mode 100644 test/account/iam_v1/test_users.py diff --git a/README.md b/README.md index 58b6c7d..618b9d7 100644 --- a/README.md +++ b/README.md @@ -327,12 +327,12 @@ asyncio.run(main()) API | Level | Access | :-----------------------------------------------------------|:------------------:|:-------------------------------------| Account Management - Account limits | :x: | | - Account Management - User management | :x: | | - Account Management - Group management | :x: | | - Account Management - Permission management | :x: | | - Account Management - Policy management | :x: | | - Account Management - Service user management | :x: | | - Account Management - Platform tokens | :x: | | + Account Management - User management | :heavy_check_mark: | `dt.account.iam_users` | + Account Management - Group management | :heavy_check_mark: | `dt.account.iam_groups` | + Account Management - Permission management | :heavy_check_mark: | `dt.account.iam_policies` | + Account Management - Policy management | :heavy_check_mark: | `dt.account.iam_policies` | + Account Management - Service user management | :heavy_check_mark: | `dt.account.iam_service_users` | + Account Management - Platform tokens | :heavy_check_mark: | `dt.account.iam_platform_tokens` | Environment management API v1 | :heavy_check_mark: | `dt.account.env_v1` | Environment management API v2 | :heavy_check_mark: | `dt.account.env_v2` | Dynatrace Platform Subscription - Subscription management | :heavy_check_mark: | `dt.account.sub_v2` | diff --git a/dynatrace/account/iam_v1/__init__.py b/dynatrace/account/iam_v1/__init__.py new file mode 100644 index 0000000..69eae2f --- /dev/null +++ b/dynatrace/account/iam_v1/__init__.py @@ -0,0 +1 @@ +"""Account IAM v1 API wrappers.""" diff --git a/dynatrace/account/iam_v1/groups.py b/dynatrace/account/iam_v1/groups.py new file mode 100644 index 0000000..0c5b469 --- /dev/null +++ b/dynatrace/account/iam_v1/groups.py @@ -0,0 +1,165 @@ +"""Account group management API wrappers.""" + +import builtins +from typing import Any, Union + +from httpx import Response + +from dynatrace.dynatrace_object import DynatraceObject +from dynatrace.http_client import HttpClient + + +class AccountGroupsService: + """ + /iam/v1 Group management API + + - GET /iam/v1/accounts/{accountUuid}/groups + - POST /iam/v1/accounts/{accountUuid}/groups + - GET /iam/v1/accounts/{accountUuid}/groups/{groupUuid}/users + - PUT /iam/v1/accounts/{accountUuid}/groups/{groupUuid} + - DELETE /iam/v1/accounts/{accountUuid}/groups/{groupUuid} + """ + + def __init__(self, http_client: HttpClient) -> None: + self.__http_client = http_client + + async def list(self, account_uuid: str) -> "GroupList": + """List all user groups of an account.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/groups" + ) + ).json() + return GroupList(raw_element=resp) + + async def create( + self, account_uuid: str, groups: builtins.list[dict[str, Any]] + ) -> builtins.list["Group"]: + """Create new user groups.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/groups", + method="POST", + json=groups, + ) + ).json() + return [Group(raw_element=e) for e in resp] + + async def get_members(self, account_uuid: str, group_uuid: str) -> "GroupMembers": + """List all members of a group.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/groups/{group_uuid}/users" + ) + ).json() + return GroupMembers(raw_element=resp) + + async def update( + self, + account_uuid: str, + group_uuid: str, + group_config: Union["GroupUpdateRequest", dict[str, Any]], + ) -> Response: + """Edit a user group.""" + if isinstance(group_config, GroupUpdateRequest): + body = group_config.to_json() + else: + body = group_config + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/groups/{group_uuid}", + method="PUT", + json=body, + ) + + async def delete(self, account_uuid: str, group_uuid: str) -> Response: + """Delete a user group.""" + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/groups/{group_uuid}", + method="DELETE", + ) + + +# Response models +class Group(DynatraceObject): + """Group object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.uuid: str | None = raw_element.get("uuid") + self.name: str | None = raw_element.get("name") + self.description: str | None = raw_element.get("description") + self.federated_attribute_values: list[str] = raw_element.get( + "federatedAttributeValues", [] + ) + self.owner: str | None = raw_element.get("owner") + self.created_at: str | None = raw_element.get("createdAt") + self.updated_at: str | None = raw_element.get("updatedAt") + + def to_json(self) -> dict[str, Any]: + return { + "uuid": self.uuid, + "name": self.name, + "description": self.description, + "federatedAttributeValues": self.federated_attribute_values, + } + + +class GroupList(DynatraceObject): + """List of groups.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.count: int | None = raw_element.get("count") + self.items: list[Group] = [ + Group(raw_element=e) for e in raw_element.get("items", []) + ] + + +class GroupUser(DynatraceObject): + """User in a group.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.uid: str | None = raw_element.get("uid") + self.email: str | None = raw_element.get("email") + self.name: str | None = raw_element.get("name") + self.surname: str | None = raw_element.get("surname") + self.user_status: str | None = raw_element.get("userStatus") + self.emergency_contact: bool | None = raw_element.get("emergencyContact") + + +class GroupMembers(DynatraceObject): + """Members of a group.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.count: int | None = raw_element.get("count") + self.items: list[GroupUser] = [ + GroupUser(raw_element=e) for e in raw_element.get("items", []) + ] + + +class GroupUpdateRequest(DynatraceObject): + """Request for updating a group.""" + + def __init__( + self, + name: str, + description: str | None = None, + federated_attribute_values: list[str] | None = None, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + self.name = name + self.description = description + self.federated_attribute_values = federated_attribute_values or [] + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.name = raw_element.get("name") + self.description = raw_element.get("description") + self.federated_attribute_values = raw_element.get( + "federatedAttributeValues", [] + ) + + def to_json(self) -> dict[str, Any]: + return { + "name": self.name, + "description": self.description, + "federatedAttributeValues": self.federated_attribute_values, + } diff --git a/dynatrace/account/iam_v1/platform_tokens.py b/dynatrace/account/iam_v1/platform_tokens.py new file mode 100644 index 0000000..c28a573 --- /dev/null +++ b/dynatrace/account/iam_v1/platform_tokens.py @@ -0,0 +1,145 @@ +"""Account platform tokens management API wrappers.""" + +import builtins +from typing import Any + +from httpx import Response + +from dynatrace.dynatrace_object import DynatraceObject +from dynatrace.http_client import HttpClient + + +class AccountPlatformTokensService: + """ + /iam/v1 Platform tokens management API + + - GET /iam/v1/accounts/{accountUuid}/platform-tokens + - POST /iam/v1/accounts/{accountUuid}/platform-tokens + - DELETE /iam/v1/accounts/{accountUuid}/platform-tokens/{platformTokenId} + - PUT /iam/v1/accounts/{accountUuid}/platform-tokens/{platformTokenId}/expiration-date + - PUT /iam/v1/accounts/{accountUuid}/platform-tokens/{platformTokenId}/status + """ + + def __init__(self, http_client: HttpClient) -> None: + self.__http_client = http_client + + async def list( + self, + account_uuid: str, + search_term: str | None = None, + status: str | None = None, + page: int | None = None, + size: int | None = None, + ) -> "PlatformTokenPage": + """List all platform tokens within account.""" + params = {} + if search_term is not None: + params["searchTerm"] = search_term + if status is not None: + params["status"] = status + if page is not None: + params["page"] = page + if size is not None: + params["size"] = size + + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/platform-tokens", + params=params, + ) + ).json() + return PlatformTokenPage(raw_element=resp) + + async def create( + self, + account_uuid: str, + name: str, + scopes: builtins.list[str], + expiration_date: str | None = None, + ) -> "PlatformTokenSecret": + """Create a new platform token for user.""" + body = {"name": name, "scopes": scopes} + if expiration_date: + body["expirationDate"] = expiration_date + + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/platform-tokens", + method="POST", + json=body, + ) + ).json() + return PlatformTokenSecret(raw_element=resp) + + async def delete(self, account_uuid: str, platform_token_id: str) -> Response: + """Delete a platform token.""" + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/platform-tokens/{platform_token_id}", + method="DELETE", + ) + + async def set_expiration_date( + self, + account_uuid: str, + platform_token_id: str, + expiration_date: str, + ) -> Response: + """Update platform token expiration date.""" + body = {"expirationDate": expiration_date} + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/platform-tokens/{platform_token_id}/expiration-date", + method="PUT", + json=body, + ) + + async def set_status( + self, + account_uuid: str, + platform_token_id: str, + status: str, + ) -> Response: + """Update platform token status.""" + body = {"status": status} + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/platform-tokens/{platform_token_id}/status", + method="PUT", + json=body, + ) + + +# Response models +class PlatformToken(DynatraceObject): + """Platform token object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.token_id: str | None = raw_element.get("tokenId") + self.name: str | None = raw_element.get("name") + self.status: str | None = raw_element.get("status") + self.created_by: str | None = raw_element.get("createdBy") + self.created_at: str | None = raw_element.get("createdAt") + self.last_used_at: str | None = raw_element.get("lastUsedAt") + self.expiration_date: str | None = raw_element.get("expirationDate") + self.scopes: list[str] = raw_element.get("scopes", []) + + +class PlatformTokenPage(DynatraceObject): + """Page of platform tokens.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.items: list[PlatformToken] = [ + PlatformToken(raw_element=e) for e in raw_element.get("items", []) + ] + self.page_size: int | None = raw_element.get("pageSize") + self.total_items: int | None = raw_element.get("totalItems") + + +class PlatformTokenSecret(DynatraceObject): + """Platform token secret (returned on creation).""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.token_id: str | None = raw_element.get("tokenId") + self.secret: str | None = raw_element.get("secret") + self.name: str | None = raw_element.get("name") + self.scopes: list[str] = raw_element.get("scopes", []) + self.expiration_date: str | None = raw_element.get("expirationDate") + self.created_at: str | None = raw_element.get("createdAt") diff --git a/dynatrace/account/iam_v1/policies.py b/dynatrace/account/iam_v1/policies.py new file mode 100644 index 0000000..62362b0 --- /dev/null +++ b/dynatrace/account/iam_v1/policies.py @@ -0,0 +1,367 @@ +"""Account policy management API wrappers.""" + +import builtins +from typing import Any, Union + +from httpx import Response + +from dynatrace.dynatrace_object import DynatraceObject +from dynatrace.http_client import HttpClient + + +class AccountPoliciesService: + """ + /iam/v1 Policy management API + + - GET /iam/v1/repo/{levelType}/{levelId}/policies + - POST /iam/v1/repo/{levelType}/{levelId}/policies + - GET /iam/v1/repo/{levelType}/{levelId}/policies/{policyUuid} + - PUT /iam/v1/repo/{levelType}/{levelId}/policies/{policyUuid} + - DELETE /iam/v1/repo/{levelType}/{levelId}/policies/{policyUuid} + - GET /iam/v1/repo/{levelType}/{levelId}/policies/aggregate + - GET /iam/v1/repo/{levelType}/{levelId}/bindings + - GET /iam/v1/repo/{levelType}/{levelId}/bindings/{policyUuid} + - POST /iam/v1/repo/{levelType}/{levelId}/bindings/{policyUuid} + - PUT /iam/v1/repo/{levelType}/{levelId}/bindings/groups/{groupUuid} + - GET /iam/v1/resolution/{levelType}/{levelId}/effectivepermissions + """ + + def __init__(self, http_client: HttpClient) -> None: + self.__http_client = http_client + + async def list( + self, + level_type: str, + level_id: str, + categories: list[str] | None = None, + name: str | None = None, + ) -> "PolicyList": + """List all native policies of a level.""" + params = {} + if categories: + params["categories"] = categories + if name: + params["name"] = name + resp = ( + await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/policies", + params=params, + ) + ).json() + return PolicyList(raw_element=resp) + + async def create( + self, + level_type: str, + level_id: str, + policy_config: Union["PolicyCreateRequest", dict[str, Any]], + ) -> "Policy": + """Create a new policy.""" + if isinstance(policy_config, PolicyCreateRequest): + body = policy_config.to_json() + else: + body = policy_config + resp = ( + await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/policies", + method="POST", + json=body, + ) + ).json() + return Policy(raw_element=resp) + + async def get(self, level_type: str, level_id: str, policy_uuid: str) -> "Policy": + """Get a policy.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/policies/{policy_uuid}" + ) + ).json() + return Policy(raw_element=resp) + + async def update( + self, + level_type: str, + level_id: str, + policy_uuid: str, + policy_config: Union["PolicyCreateRequest", dict[str, Any]], + ) -> Response: + """Update a policy.""" + if isinstance(policy_config, PolicyCreateRequest): + body = policy_config.to_json() + else: + body = policy_config + return await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/policies/{policy_uuid}", + method="PUT", + json=body, + ) + + async def delete( + self, level_type: str, level_id: str, policy_uuid: str, force: bool = False + ) -> Response: + """Delete a policy.""" + return await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/policies/{policy_uuid}", + method="DELETE", + params={"force": force}, + ) + + async def list_aggregate( + self, level_type: str, level_id: str + ) -> "PolicyOverviewList": + """List all policies for a level, including inherited from higher levels.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/policies/aggregate" + ) + ).json() + return PolicyOverviewList(raw_element=resp) + + async def list_bindings(self, level_type: str, level_id: str) -> "PolicyBindings": + """List all policy bindings of a level.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/bindings" + ) + ).json() + return PolicyBindings(raw_element=resp) + + async def get_policy_bindings( + self, level_type: str, level_id: str, policy_uuid: str + ) -> "PolicyBindings": + """Get policy bindings within a level.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/bindings/{policy_uuid}" + ) + ).json() + return PolicyBindings(raw_element=resp) + + async def add_policy_bindings( + self, + level_type: str, + level_id: str, + policy_uuid: str, + bindings_config: Union["PolicyBindingsCreateRequest", dict[str, Any]], + ) -> Response: + """Add policy bindings to a level.""" + if isinstance(bindings_config, PolicyBindingsCreateRequest): + body = bindings_config.to_json() + else: + body = bindings_config + return await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/bindings/{policy_uuid}", + method="POST", + json=body, + ) + + async def get_group_policies( + self, level_type: str, level_id: str, group_uuid: str + ) -> "PolicyUuids": + """List all policies for a user group.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/bindings/groups/{group_uuid}" + ) + ).json() + return PolicyUuids(raw_element=resp) + + async def set_group_policies( + self, + level_type: str, + level_id: str, + group_uuid: str, + policy_uuids: builtins.list[str], + ) -> Response: + """Update policy bindings for a user group (overwrites existing).""" + body = {"policyUuids": policy_uuids} + return await self.__http_client.make_request( + f"/iam/v1/repo/{level_type}/{level_id}/bindings/groups/{group_uuid}", + method="PUT", + json=body, + ) + + async def get_effective_permissions( + self, + level_type: str, + level_id: str, + entity_id: str, + entity_type: str, + services: builtins.list[str] | None = None, + page: int = 1, + size: int = 100, + ) -> "EffectivePermissions": + """Get effective permissions for a user or group.""" + params = { + "entityId": entity_id, + "entityType": entity_type, + "page": page, + "size": size, + } + if services: + params["services"] = services + resp = ( + await self.__http_client.make_request( + f"/iam/v1/resolution/{level_type}/{level_id}/effectivepermissions", + params=params, + ) + ).json() + return EffectivePermissions(raw_element=resp) + + +# Response models +class PolicyStatement(DynatraceObject): + """Policy statement.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.effect: str | None = raw_element.get("effect") + self.permissions: list[str] = raw_element.get("permissions", []) + self.conditions: list[dict[str, Any]] = raw_element.get("conditions", []) + + +class Policy(DynatraceObject): + """Policy object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.uuid: str | None = raw_element.get("uuid") + self.name: str | None = raw_element.get("name") + self.description: str | None = raw_element.get("description") + self.tags: list[str] = raw_element.get("tags", []) + self.statement_query: str | None = raw_element.get("statementQuery") + statements_data = raw_element.get("statements", []) + self.statements: list[PolicyStatement] = [ + PolicyStatement(raw_element=s) for s in statements_data + ] + + +class PolicyList(DynatraceObject): + """List of policies.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.policies: list[dict[str, Any]] = raw_element.get("policies", []) + + +class PolicyOverview(DynatraceObject): + """Policy overview object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.uuid: str | None = raw_element.get("uuid") + self.name: str | None = raw_element.get("name") + self.description: str | None = raw_element.get("description") + self.level_id: str | None = raw_element.get("levelId") + self.level_type: str | None = raw_element.get("levelType") + + +class PolicyOverviewList(DynatraceObject): + """List of policy overviews.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.policy_overview_list: list[PolicyOverview] = [ + PolicyOverview(raw_element=e) + for e in raw_element.get("policyOverviewList", []) + ] + + +class PolicyBinding(DynatraceObject): + """Policy binding object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.policy_uuid: str | None = raw_element.get("policyUuid") + self.groups: list[str] = raw_element.get("groups", []) + self.parameters: dict[str, str] = raw_element.get("parameters", {}) + self.metadata: dict[str, str] = raw_element.get("metadata", {}) + self.boundaries: list[str] = raw_element.get("boundaries", []) + + +class PolicyBindings(DynatraceObject): + """Policy bindings for a level.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.level_type: str | None = raw_element.get("levelType") + self.level_id: str | None = raw_element.get("levelId") + self.policy_bindings: list[PolicyBinding] = [ + PolicyBinding(raw_element=e) for e in raw_element.get("policyBindings", []) + ] + + +class PolicyUuids(DynatraceObject): + """List of policy UUIDs.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.policy_uuids: list[str] = raw_element.get("policyUuids", []) + + +class EffectivePermission(DynatraceObject): + """Effective permission.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.permission: str | None = raw_element.get("permission") + self.effects: list[dict[str, Any]] = raw_element.get("effects", []) + + +class EffectivePermissions(DynatraceObject): + """Effective permissions for a user or group.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.effective_permissions: list[EffectivePermission] = [ + EffectivePermission(raw_element=e) + for e in raw_element.get("effectivePermissions", []) + ] + + +# Request models +class PolicyCreateRequest(DynatraceObject): + """Request for creating a policy.""" + + def __init__( + self, + name: str, + description: str, + statement_query: str, + tags: list[str] | None = None, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + self.name = name + self.description = description + self.statement_query = statement_query + self.tags = tags or [] + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.name = raw_element.get("name") + self.description = raw_element.get("description") + self.statement_query = raw_element.get("statementQuery") + self.tags = raw_element.get("tags", []) + + def to_json(self) -> dict[str, Any]: + return { + "name": self.name, + "description": self.description, + "statementQuery": self.statement_query, + "tags": self.tags, + } + + +class PolicyBindingsCreateRequest(DynatraceObject): + """Request for creating policy bindings.""" + + def __init__( + self, + groups: list[str], + boundaries: list[str] | None = None, + **kwargs: Any, + ) -> None: + super().__init__(**kwargs) + self.groups = groups + self.boundaries = boundaries or [] + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.groups = raw_element.get("groups", []) + self.boundaries = raw_element.get("boundaries", []) + + def to_json(self) -> dict[str, Any]: + return { + "groups": self.groups, + "boundaries": self.boundaries, + } diff --git a/dynatrace/account/iam_v1/service_users.py b/dynatrace/account/iam_v1/service_users.py new file mode 100644 index 0000000..7aee8e9 --- /dev/null +++ b/dynatrace/account/iam_v1/service_users.py @@ -0,0 +1,127 @@ +"""Account service users management API wrappers.""" + +from typing import Any + +from httpx import Response + +from dynatrace.dynatrace_object import DynatraceObject +from dynatrace.http_client import HttpClient + + +class AccountServiceUsersService: + """ + /iam/v1 Service users management API + + - POST /iam/v1/accounts/{accountUuid}/service-users + - GET /iam/v1/accounts/{accountUuid}/service-users + - GET /iam/v1/accounts/{accountUuid}/service-users/{userUuid} + - PUT /iam/v1/accounts/{accountUuid}/service-users/{userUuid} + - DELETE /iam/v1/accounts/{accountUuid}/service-users/{userUuid} + """ + + def __init__(self, http_client: HttpClient) -> None: + self.__http_client = http_client + + async def create( + self, + account_uuid: str, + name: str, + description: str | None = None, + ) -> "ServiceUser": + """Create a new service user in an account.""" + body = {"name": name} + if description: + body["description"] = description + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/service-users", + method="POST", + json=body, + ) + ).json() + return ServiceUser(raw_element=resp) + + async def list( + self, + account_uuid: str, + page: int | None = None, + page_size: int | None = None, + page_key: str | None = None, + ) -> "ServiceUsersPage": + """List all service users assigned to account.""" + params = {} + if page is not None: + params["page"] = page + if page_size is not None: + params["page-size"] = page_size + if page_key is not None: + params["page-key"] = page_key + + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/service-users", + params=params, + ) + ).json() + return ServiceUsersPage(raw_element=resp) + + async def get(self, account_uuid: str, user_uuid: str) -> "ServiceUser": + """Get service user by uuid.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/service-users/{user_uuid}" + ) + ).json() + return ServiceUser(raw_element=resp) + + async def update( + self, + account_uuid: str, + user_uuid: str, + name: str | None = None, + description: str | None = None, + ) -> Response: + """Update name and description of service user in an account.""" + body = {} + if name is not None: + body["name"] = name + if description is not None: + body["description"] = description + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/service-users/{user_uuid}", + method="PUT", + json=body, + ) + + async def delete(self, account_uuid: str, user_uuid: str) -> Response: + """Delete a service user from an account.""" + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/service-users/{user_uuid}", + method="DELETE", + ) + + +# Response models +class ServiceUser(DynatraceObject): + """Service user object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.user_uuid: str | None = raw_element.get("userUuid") + self.name: str | None = raw_element.get("name") + self.description: str | None = raw_element.get("description") + self.created_at: str | None = raw_element.get("createdAt") + self.created_by: str | None = raw_element.get("createdBy") + self.updated_at: str | None = raw_element.get("updatedAt") + self.updated_by: str | None = raw_element.get("updatedBy") + self.group_uuids: list[str] = raw_element.get("groupUuids", []) + + +class ServiceUsersPage(DynatraceObject): + """Page of service users.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.items: list[ServiceUser] = [ + ServiceUser(raw_element=e) for e in raw_element.get("items", []) + ] + self.next_page_key: str | None = raw_element.get("nextPageKey") + self.page_size: int | None = raw_element.get("pageSize") diff --git a/dynatrace/account/iam_v1/users.py b/dynatrace/account/iam_v1/users.py new file mode 100644 index 0000000..eb69530 --- /dev/null +++ b/dynatrace/account/iam_v1/users.py @@ -0,0 +1,176 @@ +"""Account user management API wrappers.""" + +import builtins +from typing import Any + +from httpx import Response + +from dynatrace.dynatrace_object import DynatraceObject +from dynatrace.http_client import HttpClient + + +class AccountUsersService: + """ + /iam/v1 User management API + + - GET /iam/v1/accounts/{accountUuid}/users + - POST /iam/v1/accounts/{accountUuid}/users + - GET /iam/v1/accounts/{accountUuid}/users/{email} + - POST /iam/v1/accounts/{accountUuid}/users/{email} + - PUT /iam/v1/accounts/{accountUuid}/users/{email}/groups + - DELETE /iam/v1/accounts/{accountUuid}/users/{email}/groups + - DELETE /iam/v1/accounts/{accountUuid}/users/{email} + """ + + def __init__(self, http_client: HttpClient) -> None: + self.__http_client = http_client + + async def list(self, account_uuid: str, service_users: bool = False) -> "UserList": + """List all users of an account.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/users", + params={"service-users": service_users}, + ) + ).json() + return UserList(raw_element=resp) + + async def create(self, account_uuid: str, email: str) -> "UserCreationResponse": + """Create a new user in an account.""" + body = {"email": email} + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/users", + method="POST", + json=body, + ) + ).json() + return UserCreationResponse(raw_element=resp) + + async def get_groups(self, account_uuid: str, email: str) -> "UserGroups": + """List all groups of a user.""" + resp = ( + await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/users/{email}" + ) + ).json() + return UserGroups(raw_element=resp) + + async def add_to_groups( + self, account_uuid: str, email: str, group_uuids: builtins.list[str] + ) -> Response: + """Add a user to groups.""" + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/users/{email}", + method="POST", + json=group_uuids, + ) + + async def set_groups( + self, account_uuid: str, email: str, group_uuids: builtins.list[str] + ) -> Response: + """Set group membership of a user (overwrites existing).""" + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/users/{email}/groups", + method="PUT", + json=group_uuids, + ) + + async def remove_from_groups( + self, + account_uuid: str, + email: str, + group_uuids: builtins.list[str], + ) -> Response: + """Remove a user from groups.""" + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/users/{email}/groups", + method="DELETE", + params={"group-uuid": group_uuids}, + ) + + async def delete(self, account_uuid: str, email: str) -> Response: + """Remove a user from an account.""" + return await self.__http_client.make_request( + f"/iam/v1/accounts/{account_uuid}/users/{email}", + method="DELETE", + ) + + +# Response models +class UserLoginMetadata(DynatraceObject): + """User login metadata.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.successful_login_counter: int | None = raw_element.get( + "successfulLoginCounter" + ) + self.failed_login_counter: int | None = raw_element.get("failedLoginCounter") + self.last_successful_login: str | None = raw_element.get("lastSuccessfulLogin") + self.last_failed_login: str | None = raw_element.get("lastFailedLogin") + self.created_at: str | None = raw_element.get("createdAt") + self.updated_at: str | None = raw_element.get("updatedAt") + + +class User(DynatraceObject): + """User object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.uid: str | None = raw_element.get("uid") + self.email: str | None = raw_element.get("email") + self.name: str | None = raw_element.get("name") + self.surname: str | None = raw_element.get("surname") + self.user_status: str | None = raw_element.get("userStatus") + self.emergency_contact: bool | None = raw_element.get("emergencyContact") + user_login_metadata = raw_element.get("userLoginMetadata") + self.user_login_metadata: UserLoginMetadata | None = ( + UserLoginMetadata(raw_element=user_login_metadata) + if user_login_metadata + else None + ) + + +class UserList(DynatraceObject): + """List of users.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.count: int | None = raw_element.get("count") + self.items: list[User] = [ + User(raw_element=e) for e in raw_element.get("items", []) + ] + + +class UserCreationResponse(DynatraceObject): + """Response from user creation.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.user_uuid: str | None = raw_element.get("userUuid") + + +class Group(DynatraceObject): + """Group object.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.group_name: str | None = raw_element.get("groupName") + self.uuid: str | None = raw_element.get("uuid") + self.owner: str | None = raw_element.get("owner") + self.account_uuid: str | None = raw_element.get("accountUUID") + self.account_name: str | None = raw_element.get("accountName") + self.description: str | None = raw_element.get("description") + self.created_at: str | None = raw_element.get("createdAt") + self.updated_at: str | None = raw_element.get("updatedAt") + + +class UserGroups(DynatraceObject): + """User with their groups.""" + + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.uid: str | None = raw_element.get("uid") + self.email: str | None = raw_element.get("email") + self.name: str | None = raw_element.get("name") + self.surname: str | None = raw_element.get("surname") + self.user_status: str | None = raw_element.get("userStatus") + self.emergency_contact: bool | None = raw_element.get("emergencyContact") + self.groups: list[Group] = [ + Group(raw_element=e) for e in raw_element.get("groups", []) + ] diff --git a/dynatrace/account/schemas.py b/dynatrace/account/schemas.py index 1445083..9b92eef 100644 --- a/dynatrace/account/schemas.py +++ b/dynatrace/account/schemas.py @@ -3,6 +3,11 @@ from dynatrace.account.env_v1.environments import AccountEnvironmentsV1Service from dynatrace.account.env_v2.environments import AccountEnvironmentsV2Service from dynatrace.account.env_v2.settings import SettingService as AccountSettingService +from dynatrace.account.iam_v1.groups import AccountGroupsService +from dynatrace.account.iam_v1.platform_tokens import AccountPlatformTokensService +from dynatrace.account.iam_v1.policies import AccountPoliciesService +from dynatrace.account.iam_v1.service_users import AccountServiceUsersService +from dynatrace.account.iam_v1.users import AccountUsersService from dynatrace.account.sub_v1.cost_allocation import CostAllocationService from dynatrace.account.sub_v1.rate_cards import RateCardService from dynatrace.account.sub_v2.subscriptions import SubscriptionService @@ -21,6 +26,17 @@ def __init__(self, http_client: HttpClient) -> None: ) self.settings: AccountSettingService = AccountSettingService(http_client) + # Account IAM APIs. + self.iam_users: AccountUsersService = AccountUsersService(http_client) + self.iam_groups: AccountGroupsService = AccountGroupsService(http_client) + self.iam_policies: AccountPoliciesService = AccountPoliciesService(http_client) + self.iam_service_users: AccountServiceUsersService = AccountServiceUsersService( + http_client + ) + self.iam_platform_tokens: AccountPlatformTokensService = ( + AccountPlatformTokensService(http_client) + ) + # Account subscription APIs. self.sub_v1_cost_allocation: CostAllocationService = CostAllocationService( http_client diff --git a/pyproject.toml b/pyproject.toml index b47329d..37fa7c5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "dt-async" -version = "1.0.7" +version = "1.0.8" description = "Dynatrace async API Python client" readme = "README.md" authors = ["David Lopes "] diff --git a/test/account/iam_v1/__init__.py b/test/account/iam_v1/__init__.py new file mode 100644 index 0000000..2f5a6ac --- /dev/null +++ b/test/account/iam_v1/__init__.py @@ -0,0 +1 @@ +"""IAM v1 tests init file.""" diff --git a/test/account/iam_v1/test_groups.py b/test/account/iam_v1/test_groups.py new file mode 100644 index 0000000..c1f6ad2 --- /dev/null +++ b/test/account/iam_v1/test_groups.py @@ -0,0 +1,225 @@ +"""Tests for account IAM v1 groups API.""" + +import json +from unittest import mock + +from dynatrace import DynatraceAsync +from dynatrace.account.iam_v1.groups import ( + AccountGroupsService, + Group, + GroupList, + GroupMembers, + GroupUpdateRequest, +) +from dynatrace.http_client import HttpClient + + +class MockResponse: + def __init__(self, json_data=None, status_code=200): + self._json_data = json_data + self.headers = {} + self.text = json.dumps(json_data) if json_data is not None else "" + self.status_code = status_code + + def json(self): + return self._json_data + + +async def test_groups_service_is_exposed(dt: DynatraceAsync): + assert isinstance(dt.account.iam_groups, AccountGroupsService) + + +async def test_groups_list(dt: DynatraceAsync): + account_uuid = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ("GET", f"/iam/v1/accounts/{account_uuid}/groups"): + return MockResponse( + { + "count": 1, + "items": [ + { + "uuid": "group-1", + "name": "Admins", + "description": "Admin group", + "federatedAttributeValues": [], + "owner": "LOCAL", + "createdAt": "2021-05-01T15:11:00Z", + "updatedAt": "2021-05-01T15:11:00Z", + } + ], + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + groups = await dt.account.iam_groups.list(account_uuid) + + assert isinstance(groups, GroupList) + assert groups.count == 1 + assert len(groups.items) == 1 + assert groups.items[0].name == "Admins" + + +async def test_groups_create(dt: DynatraceAsync): + account_uuid = "account-123" + group_config = [{"name": "TestGroup", "description": "Test Group"}] + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if (method, path) == ("POST", f"/iam/v1/accounts/{account_uuid}/groups"): + assert json == group_config + return MockResponse( + [ + { + "uuid": "new-group-1", + "name": "TestGroup", + "description": "Test Group", + "federatedAttributeValues": [], + "owner": "LOCAL", + "createdAt": "2021-05-01T15:11:00Z", + "updatedAt": "2021-05-01T15:11:00Z", + } + ], + 201, + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + groups = await dt.account.iam_groups.create(account_uuid, group_config) + + assert len(groups) == 1 + assert isinstance(groups[0], Group) + assert groups[0].name == "TestGroup" + + +async def test_groups_get_members(dt: DynatraceAsync): + account_uuid = "account-123" + group_uuid = "group-1" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if ( + method, + path, + ) == ("GET", f"/iam/v1/accounts/{account_uuid}/groups/{group_uuid}/users"): + return MockResponse( + { + "count": 1, + "items": [ + { + "uid": "user-1", + "email": "user@example.com", + "name": "John", + "surname": "Doe", + "userStatus": "ACTIVE", + "emergencyContact": False, + } + ], + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + members = await dt.account.iam_groups.get_members(account_uuid, group_uuid) + + assert isinstance(members, GroupMembers) + assert members.count == 1 + assert len(members.items) == 1 + assert members.items[0].email == "user@example.com" + + +async def test_groups_update(dt: DynatraceAsync): + account_uuid = "account-123" + group_uuid = "group-1" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if ( + method, + path, + ) == ("PUT", f"/iam/v1/accounts/{account_uuid}/groups/{group_uuid}"): + assert json == { + "name": "Updated Group", + "description": "Updated description", + "federatedAttributeValues": [], + } + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + group_config = GroupUpdateRequest( + name="Updated Group", description="Updated description" + ) + response = await dt.account.iam_groups.update( + account_uuid, group_uuid, group_config + ) + + assert response.status_code == 200 + + +async def test_groups_delete(dt: DynatraceAsync): + account_uuid = "account-123" + group_uuid = "group-1" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if ( + method, + path, + ) == ("DELETE", f"/iam/v1/accounts/{account_uuid}/groups/{group_uuid}"): + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_groups.delete(account_uuid, group_uuid) + + assert response.status_code == 200 diff --git a/test/account/iam_v1/test_platform_tokens.py b/test/account/iam_v1/test_platform_tokens.py new file mode 100644 index 0000000..fbfc21c --- /dev/null +++ b/test/account/iam_v1/test_platform_tokens.py @@ -0,0 +1,219 @@ +"""Tests for account IAM v1 platform tokens API.""" + +import json +from unittest import mock + +from dynatrace import DynatraceAsync +from dynatrace.account.iam_v1.platform_tokens import ( + AccountPlatformTokensService, + PlatformTokenPage, + PlatformTokenSecret, +) +from dynatrace.http_client import HttpClient + + +class MockResponse: + def __init__(self, json_data=None, status_code=200): + self._json_data = json_data + self.headers = {} + self.text = json.dumps(json_data) if json_data is not None else "" + self.status_code = status_code + + def json(self): + return self._json_data + + +async def test_platform_tokens_service_is_exposed(dt: DynatraceAsync): + assert isinstance(dt.account.iam_platform_tokens, AccountPlatformTokensService) + + +async def test_platform_tokens_list(dt: DynatraceAsync): + account_uuid = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/accounts/{account_uuid}/platform-tokens", + ): + return MockResponse( + { + "items": [ + { + "tokenId": "token-123", + "name": "Test Token", + "status": "ACTIVE", + "createdBy": "user@example.com", + "createdAt": "2021-05-01T15:11:00Z", + "lastUsedAt": "2021-05-02T10:00:00Z", + "expirationDate": "2025-05-01T15:11:00Z", + "scopes": ["platform-token:tokens:manage"], + } + ], + "pageSize": 1, + "totalItems": 1, + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + page = await dt.account.iam_platform_tokens.list(account_uuid) + + assert isinstance(page, PlatformTokenPage) + assert len(page.items) == 1 + assert page.items[0].name == "Test Token" + assert page.items[0].status == "ACTIVE" + + +async def test_platform_tokens_create(dt: DynatraceAsync): + account_uuid = "account-123" + name = "New Token" + scopes = ["platform-token:tokens:manage"] + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if (method, path) == ( + "POST", + f"/iam/v1/accounts/{account_uuid}/platform-tokens", + ): + assert json == {"name": name, "scopes": scopes} + return MockResponse( + { + "tokenId": "token-456", + "secret": "secret_token_value", + "name": name, + "scopes": scopes, + "createdAt": "2021-05-01T15:11:00Z", + }, + 201, + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + token = await dt.account.iam_platform_tokens.create(account_uuid, name, scopes) + + assert isinstance(token, PlatformTokenSecret) + assert token.token_id == "token-456" + assert token.secret == "secret_token_value" + + +async def test_platform_tokens_delete(dt: DynatraceAsync): + account_uuid = "account-123" + token_id = "token-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if ( + method, + path, + ) == ( + "DELETE", + f"/iam/v1/accounts/{account_uuid}/platform-tokens/{token_id}", + ): + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_platform_tokens.delete(account_uuid, token_id) + + assert response.status_code == 200 + + +async def test_platform_tokens_set_expiration_date(dt: DynatraceAsync): + account_uuid = "account-123" + token_id = "token-123" + expiration_date = "2026-05-01T00:00:00Z" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if ( + method, + path, + ) == ( + "PUT", + f"/iam/v1/accounts/{account_uuid}/platform-tokens/{token_id}/expiration-date", + ): + assert json == {"expirationDate": expiration_date} + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_platform_tokens.set_expiration_date( + account_uuid, token_id, expiration_date + ) + + assert response.status_code == 200 + + +async def test_platform_tokens_set_status(dt: DynatraceAsync): + account_uuid = "account-123" + token_id = "token-123" + status = "INACTIVE" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if ( + method, + path, + ) == ( + "PUT", + f"/iam/v1/accounts/{account_uuid}/platform-tokens/{token_id}/status", + ): + assert json == {"status": status} + return MockResponse({}, 204) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_platform_tokens.set_status( + account_uuid, token_id, status + ) + + assert response.status_code == 204 diff --git a/test/account/iam_v1/test_policies.py b/test/account/iam_v1/test_policies.py new file mode 100644 index 0000000..ae9c578 --- /dev/null +++ b/test/account/iam_v1/test_policies.py @@ -0,0 +1,346 @@ +"""Tests for account IAM v1 policies API.""" + +import json +from unittest import mock + +from dynatrace import DynatraceAsync +from dynatrace.account.iam_v1.policies import ( + AccountPoliciesService, + EffectivePermissions, + Policy, + PolicyBindings, + PolicyCreateRequest, + PolicyList, + PolicyOverviewList, +) +from dynatrace.http_client import HttpClient + + +class MockResponse: + def __init__(self, json_data=None, status_code=200): + self._json_data = json_data + self.headers = {} + self.text = json.dumps(json_data) if json_data is not None else "" + self.status_code = status_code + + def json(self): + return self._json_data + + +async def test_policies_service_is_exposed(dt: DynatraceAsync): + assert isinstance(dt.account.iam_policies, AccountPoliciesService) + + +async def test_policies_list(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/repo/{level_type}/{level_id}/policies", + ): + return MockResponse( + { + "policies": [ + { + "uuid": "policy-1", + "name": "TestPolicy", + "description": "Test Policy", + } + ] + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + policies = await dt.account.iam_policies.list(level_type, level_id) + + assert isinstance(policies, PolicyList) + assert len(policies.policies) == 1 + + +async def test_policies_create(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if (method, path) == ( + "POST", + f"/iam/v1/repo/{level_type}/{level_id}/policies", + ): + return MockResponse( + { + "uuid": "new-policy-1", + "name": "NewPolicy", + "description": "New Policy", + "tags": [], + "statementQuery": "ALLOW action", + "statements": [], + }, + 201, + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + policy_config = PolicyCreateRequest( + name="NewPolicy", + description="New Policy", + statement_query="ALLOW action", + ) + policy = await dt.account.iam_policies.create( + level_type, level_id, policy_config + ) + + assert isinstance(policy, Policy) + assert policy.name == "NewPolicy" + + +async def test_policies_get(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + policy_uuid = "policy-1" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/repo/{level_type}/{level_id}/policies/{policy_uuid}", + ): + return MockResponse( + { + "uuid": policy_uuid, + "name": "TestPolicy", + "description": "Test Policy", + "tags": [], + "statementQuery": "ALLOW action", + "statements": [], + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + policy = await dt.account.iam_policies.get(level_type, level_id, policy_uuid) + + assert isinstance(policy, Policy) + assert policy.uuid == policy_uuid + + +async def test_policies_update(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + policy_uuid = "policy-1" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if ( + method, + path, + ) == ( + "PUT", + f"/iam/v1/repo/{level_type}/{level_id}/policies/{policy_uuid}", + ): + return MockResponse({}, 204) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + policy_config = PolicyCreateRequest( + name="UpdatedPolicy", + description="Updated Policy", + statement_query="ALLOW action", + ) + response = await dt.account.iam_policies.update( + level_type, level_id, policy_uuid, policy_config + ) + + assert response.status_code == 204 + + +async def test_policies_delete(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + policy_uuid = "policy-1" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if ( + method, + path, + ) == ( + "DELETE", + f"/iam/v1/repo/{level_type}/{level_id}/policies/{policy_uuid}", + ): + return MockResponse({}, 204) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_policies.delete( + level_type, level_id, policy_uuid + ) + + assert response.status_code == 204 + + +async def test_policies_list_aggregate(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/repo/{level_type}/{level_id}/policies/aggregate", + ): + return MockResponse( + { + "policyOverviewList": [ + { + "uuid": "policy-1", + "name": "TestPolicy", + "description": "Test Policy", + "levelId": level_id, + "levelType": level_type, + } + ] + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + policies = await dt.account.iam_policies.list_aggregate(level_type, level_id) + + assert isinstance(policies, PolicyOverviewList) + assert len(policies.policy_overview_list) == 1 + + +async def test_policies_list_bindings(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/repo/{level_type}/{level_id}/bindings", + ): + return MockResponse( + { + "levelType": level_type, + "levelId": level_id, + "policyBindings": [ + { + "policyUuid": "policy-1", + "groups": ["group-1"], + "parameters": {}, + "metadata": {}, + "boundaries": [], + } + ], + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + bindings = await dt.account.iam_policies.list_bindings(level_type, level_id) + + assert isinstance(bindings, PolicyBindings) + assert len(bindings.policy_bindings) == 1 + + +async def test_policies_get_effective_permissions(dt: DynatraceAsync): + level_type = "account" + level_id = "account-123" + entity_id = "user-1" + entity_type = "user" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/resolution/{level_type}/{level_id}/effectivepermissions", + ): + return MockResponse({"effectivePermissions": []}) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + permissions = await dt.account.iam_policies.get_effective_permissions( + level_type, level_id, entity_id, entity_type + ) + + assert isinstance(permissions, EffectivePermissions) diff --git a/test/account/iam_v1/test_service_users.py b/test/account/iam_v1/test_service_users.py new file mode 100644 index 0000000..df738f6 --- /dev/null +++ b/test/account/iam_v1/test_service_users.py @@ -0,0 +1,226 @@ +"""Tests for account IAM v1 service users API.""" + +import json +from unittest import mock + +from dynatrace import DynatraceAsync +from dynatrace.account.iam_v1.service_users import ( + AccountServiceUsersService, + ServiceUser, + ServiceUsersPage, +) +from dynatrace.http_client import HttpClient + + +class MockResponse: + def __init__(self, json_data=None, status_code=200): + self._json_data = json_data + self.headers = {} + self.text = json.dumps(json_data) if json_data is not None else "" + self.status_code = status_code + + def json(self): + return self._json_data + + +async def test_service_users_service_is_exposed(dt: DynatraceAsync): + assert isinstance(dt.account.iam_service_users, AccountServiceUsersService) + + +async def test_service_users_create(dt: DynatraceAsync): + account_uuid = "account-123" + name = "test-service-user" + description = "Test Service User" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if (method, path) == ( + "POST", + f"/iam/v1/accounts/{account_uuid}/service-users", + ): + assert json == {"name": name, "description": description} + return MockResponse( + { + "userUuid": "service-user-123", + "name": name, + "description": description, + "createdAt": "2021-05-01T15:11:00Z", + "createdBy": "user@example.com", + "updatedAt": "2021-05-01T15:11:00Z", + "updatedBy": "user@example.com", + "groupUuids": [], + }, + 201, + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + service_user = await dt.account.iam_service_users.create( + account_uuid, name, description + ) + + assert isinstance(service_user, ServiceUser) + assert service_user.user_uuid == "service-user-123" + assert service_user.name == name + + +async def test_service_users_list(dt: DynatraceAsync): + account_uuid = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/accounts/{account_uuid}/service-users", + ): + return MockResponse( + { + "items": [ + { + "userUuid": "service-user-1", + "name": "Test Service User", + "description": "Test", + "createdAt": "2021-05-01T15:11:00Z", + "createdBy": "user@example.com", + "updatedAt": "2021-05-01T15:11:00Z", + "updatedBy": "user@example.com", + "groupUuids": [], + } + ], + "pageSize": 1, + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + page = await dt.account.iam_service_users.list(account_uuid) + + assert isinstance(page, ServiceUsersPage) + assert len(page.items) == 1 + assert page.items[0].name == "Test Service User" + + +async def test_service_users_get(dt: DynatraceAsync): + account_uuid = "account-123" + user_uuid = "service-user-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ( + "GET", + f"/iam/v1/accounts/{account_uuid}/service-users/{user_uuid}", + ): + return MockResponse( + { + "userUuid": user_uuid, + "name": "Test Service User", + "description": "Test", + "createdAt": "2021-05-01T15:11:00Z", + "createdBy": "user@example.com", + "updatedAt": "2021-05-01T15:11:00Z", + "updatedBy": "user@example.com", + "groupUuids": [], + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + service_user = await dt.account.iam_service_users.get(account_uuid, user_uuid) + + assert isinstance(service_user, ServiceUser) + assert service_user.user_uuid == user_uuid + + +async def test_service_users_update(dt: DynatraceAsync): + account_uuid = "account-123" + user_uuid = "service-user-123" + new_name = "Updated Service User" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if ( + method, + path, + ) == ( + "PUT", + f"/iam/v1/accounts/{account_uuid}/service-users/{user_uuid}", + ): + assert json == {"name": new_name} + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_service_users.update( + account_uuid, user_uuid, name=new_name + ) + + assert response.status_code == 200 + + +async def test_service_users_delete(dt: DynatraceAsync): + account_uuid = "account-123" + user_uuid = "service-user-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if ( + method, + path, + ) == ( + "DELETE", + f"/iam/v1/accounts/{account_uuid}/service-users/{user_uuid}", + ): + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_service_users.delete(account_uuid, user_uuid) + + assert response.status_code == 200 diff --git a/test/account/iam_v1/test_users.py b/test/account/iam_v1/test_users.py new file mode 100644 index 0000000..f15e6c4 --- /dev/null +++ b/test/account/iam_v1/test_users.py @@ -0,0 +1,244 @@ +"""Tests for account IAM v1 users API.""" + +import json +from unittest import mock + +from dynatrace import DynatraceAsync +from dynatrace.account.iam_v1.users import ( + AccountUsersService, + UserGroups, + UserList, +) +from dynatrace.http_client import HttpClient + + +class MockResponse: + def __init__(self, json_data=None, status_code=200): + self._json_data = json_data + self.headers = {} + self.text = json.dumps(json_data) if json_data is not None else "" + self.status_code = status_code + + def json(self): + return self._json_data + + +async def test_users_service_is_exposed(dt: DynatraceAsync): + assert isinstance(dt.account.iam_users, AccountUsersService) + + +async def test_users_list_returns_user_list(dt: DynatraceAsync): + account_uuid = "account-123" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ("GET", f"/iam/v1/accounts/{account_uuid}/users"): + return MockResponse( + { + "count": 2, + "items": [ + { + "uid": "user-1", + "email": "user1@example.com", + "name": "John", + "surname": "Doe", + "userStatus": "ACTIVE", + "emergencyContact": False, + }, + { + "uid": "user-2", + "email": "user2@example.com", + "name": "Jane", + "surname": "Smith", + "userStatus": "ACTIVE", + "emergencyContact": True, + }, + ], + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + users = await dt.account.iam_users.list(account_uuid) + + assert isinstance(users, UserList) + assert users.count == 2 + assert len(users.items) == 2 + assert users.items[0].email == "user1@example.com" + assert users.items[1].emergency_contact is True + + +async def test_users_create(dt: DynatraceAsync): + account_uuid = "account-123" + email = "newuser@example.com" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if (method, path) == ("POST", f"/iam/v1/accounts/{account_uuid}/users"): + assert json == {"email": email} + return MockResponse({"userUuid": "new-user-uuid"}, 201) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_users.create(account_uuid, email) + + assert response.user_uuid == "new-user-uuid" + + +async def test_users_get_groups(dt: DynatraceAsync): + account_uuid = "account-123" + email = "user@example.com" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if (method, path) == ("GET", f"/iam/v1/accounts/{account_uuid}/users/{email}"): + return MockResponse( + { + "uid": "user-1", + "email": email, + "name": "John", + "surname": "Doe", + "userStatus": "ACTIVE", + "emergencyContact": False, + "groups": [ + { + "groupName": "Group1", + "uuid": "group-1", + "owner": "LOCAL", + "accountUUID": account_uuid, + "accountName": "Test Account", + "description": "Test Group", + "createdAt": "2021-05-01T15:11:00Z", + "updatedAt": "2021-05-01T15:11:00Z", + } + ], + } + ) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + user_groups = await dt.account.iam_users.get_groups(account_uuid, email) + + assert isinstance(user_groups, UserGroups) + assert user_groups.email == email + assert len(user_groups.groups) == 1 + assert user_groups.groups[0].group_name == "Group1" + + +async def test_users_add_to_groups(dt: DynatraceAsync): + account_uuid = "account-123" + email = "user@example.com" + group_uuids = ["group-1", "group-2"] + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if (method, path) == ("POST", f"/iam/v1/accounts/{account_uuid}/users/{email}"): + assert json == group_uuids + return MockResponse({}, 201) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_users.add_to_groups( + account_uuid, email, group_uuids + ) + + assert response.status_code == 201 + + +async def test_users_set_groups(dt: DynatraceAsync): + account_uuid = "account-123" + email = "user@example.com" + group_uuids = ["group-1"] + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + json=None, + **kwargs, + ): + if ( + method, + path, + ) == ("PUT", f"/iam/v1/accounts/{account_uuid}/users/{email}/groups"): + assert json == group_uuids + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_users.set_groups( + account_uuid, email, group_uuids + ) + + assert response.status_code == 200 + + +async def test_users_delete(dt: DynatraceAsync): + account_uuid = "account-123" + email = "user@example.com" + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + if ( + method, + path, + ) == ("DELETE", f"/iam/v1/accounts/{account_uuid}/users/{email}"): + return MockResponse({}, 200) + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + response = await dt.account.iam_users.delete(account_uuid, email) + + assert response.status_code == 200