From f698b5413025b62b60ac0dfc61c515acfa9c700c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Harbuz?= Date: Wed, 6 May 2026 10:53:16 +0400 Subject: [PATCH] feat: Add Account Audits and Notifications API services with tests --- README.md | 4 +- dynatrace/account/audits.py | 92 ++++++++++++++++++++++ dynatrace/account/notifications.py | 114 +++++++++++++++++++++++++++ dynatrace/account/schemas.py | 8 ++ pyproject.toml | 2 +- test/account/test_audits.py | 108 ++++++++++++++++++++++++++ test/account/test_notifications.py | 120 +++++++++++++++++++++++++++++ 7 files changed, 445 insertions(+), 3 deletions(-) create mode 100644 dynatrace/account/audits.py create mode 100644 dynatrace/account/notifications.py create mode 100644 test/account/test_audits.py create mode 100644 test/account/test_notifications.py diff --git a/README.md b/README.md index 618b9d7..6679426 100644 --- a/README.md +++ b/README.md @@ -341,6 +341,6 @@ asyncio.run(main()) Dynatrace Platform Subscription - Rate cards | :heavy_check_mark: | `dt.account.sub_v1_rate_cards` | Dynatrace Platform Subscription - Cost allocation | :heavy_check_mark: | `dt.account.sub_v1_cost_allocation` | Account Settings | :x: | | - Account Audits | :x: | | + Account Audits | :heavy_check_mark: | `dt.account.audits` | Reference data | :x: | | - Notifications | :x: | | + Notifications | :heavy_check_mark: | `dt.account.notifications` | diff --git a/dynatrace/account/audits.py b/dynatrace/account/audits.py new file mode 100644 index 0000000..8dfe470 --- /dev/null +++ b/dynatrace/account/audits.py @@ -0,0 +1,92 @@ +"""Account audits API wrappers.""" + +import builtins +from datetime import datetime +from typing import Any + +from dynatrace.dynatrace_object import DynatraceObject +from dynatrace.http_client import HttpClient +from dynatrace.utils import timestamp_to_string + + +class AccountAuditsService: + """/audit/v1 Account audits API.""" + + def __init__(self, http_client: HttpClient) -> None: + self.__http_client = http_client + + async def list( + self, + account_uuid: str, + start_time: datetime | str | None = None, + end_time: datetime | str | None = None, + add_fields: builtins.list[str] | None = None, + filter: str | None = None, + limit: int | None = None, + scan_limit_gigabyte: int | None = None, + result_size_limit_megabyte: int | None = None, + ) -> "AuditsByAccount": + params: dict[str, Any] = { + "startTime": timestamp_to_string(start_time), + "endTime": timestamp_to_string(end_time), + "filter": filter, + "limit": limit, + "scanLimitGigabyte": scan_limit_gigabyte, + "resultSizeLimitMegabyte": result_size_limit_megabyte, + } + if add_fields: + params["addFields"] = ",".join(add_fields) + + resp = ( + await self.__http_client.make_request( + f"/audit/v1/accounts/{account_uuid}", + params=params, + ) + ).json() + return AuditsByAccount(raw_element=resp) + + +class Audit(DynatraceObject): + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.event_id: str | None = raw_element.get("eventId") + self.timestamp: str | None = raw_element.get("timestamp") + self.user: str | None = raw_element.get("user") + self.resource: str | None = raw_element.get("resource") + self.resource_name: str | None = raw_element.get("resourceName") + self.event_provider: str | None = raw_element.get("eventProvider") + self.event_type: str | None = raw_element.get("eventType") + self.account_uuid: str | None = raw_element.get("accountUuid") + self.authentication_client_id: str | None = raw_element.get( + "authenticationClientId" + ) + self.authentication_grant_type: str | None = raw_element.get( + "authenticationGrantType" + ) + self.authentication_token: str | None = raw_element.get("authenticationToken") + self.authentication_type: str | None = raw_element.get("authenticationType") + self.details: dict[str, str] | None = raw_element.get("details") + self.event_outcome: str | None = raw_element.get("eventOutcome") + self.event_reason: str | None = raw_element.get("eventReason") + self.event_version: str | None = raw_element.get("eventVersion") + self.origin_address: str | None = raw_element.get("originAddress") + self.origin_session: str | None = raw_element.get("originSession") + self.origin_type: str | None = raw_element.get("originType") + self.origin_x_forwarded_for: str | None = raw_element.get("originXForwardedFor") + self.resource_id: str | None = raw_element.get("resourceId") + self.environment_uuid: str | None = raw_element.get("environmentUuid") + self.user_organization: str | None = raw_element.get("userOrganization") + + +class AuditWarning(DynatraceObject): + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.message: str | None = raw_element.get("message") + + +class AuditsByAccount(DynatraceObject): + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.audits: builtins.list[Audit] = [ + Audit(raw_element=a) for a in raw_element.get("audits", []) + ] + self.warnings: builtins.list[AuditWarning] = [ + AuditWarning(raw_element=w) for w in raw_element.get("warnings", []) + ] diff --git a/dynatrace/account/notifications.py b/dynatrace/account/notifications.py new file mode 100644 index 0000000..277adc9 --- /dev/null +++ b/dynatrace/account/notifications.py @@ -0,0 +1,114 @@ +"""Account notifications API wrappers.""" + +from __future__ import annotations + +import builtins +from datetime import datetime +from enum import Enum +from typing import Any + +from dynatrace.dynatrace_object import DynatraceObject +from dynatrace.http_client import HttpClient +from dynatrace.utils import timestamp_to_string + + +class NotificationService: + """/v1/accounts/{accountUuid}/notifications API.""" + + def __init__(self, http_client: HttpClient) -> None: + self.__http_client = http_client + + async def list( + self, + account_uuid: str, + start_date_time: datetime | str | None = None, + end_date_time: datetime | str | None = None, + types: builtins.list[str | NotificationType] | None = None, + severities: builtins.list[str | NotificationSeverity] | None = None, + capabilities: builtins.list[str] | None = None, + environments: builtins.list[str] | None = None, + page: int | None = None, + page_size: int | None = None, + sorts: builtins.list[str | NotificationSortField] | None = None, + ) -> NotificationList: + body: dict[str, Any] = { + "startDateTime": timestamp_to_string(start_date_time), + "endDateTime": timestamp_to_string(end_date_time), + "types": [self._enum_value(item) for item in types] if types else None, + "severities": ( + [self._enum_value(item) for item in severities] if severities else None + ), + "capabilities": capabilities, + "environments": environments, + "page": page, + "pageSize": page_size, + "sorts": [self._enum_value(item) for item in sorts] if sorts else None, + } + + resp = ( + await self.__http_client.make_request( + f"/v1/accounts/{account_uuid}/notifications", + method="POST", + params=body, + ) + ).json() + return NotificationList(raw_element=resp) + + @staticmethod + def _enum_value(value: str | Enum) -> str: + return value.value if isinstance(value, Enum) else value + + +class NotificationType(Enum): + FORECAST = "FORECAST" + BUDGET = "BUDGET" + COST = "COST" + BYOK_REVOKED = "BYOK_REVOKED" + BYOK_ACTIVATED = "BYOK_ACTIVATED" + + +class NotificationSeverity(Enum): + SEVERE = "SEVERE" + WARN = "WARN" + INFO = "INFO" + + +class NotificationSortField(Enum): + TYPE = "type" + TYPE_DESC = "-type" + DATE = "date" + DATE_DESC = "-date" + + +class NotificationDetails(DynatraceObject): + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.environments: builtins.list[str] | None = raw_element.get("environments") + self.capabilities: builtins.list[str] | None = raw_element.get("capabilities") + self.all_environments: bool | None = raw_element.get("allEnvironments") + self.all_capabilities: bool | None = raw_element.get("allCapabilities") + self.environment_uuid: str | None = raw_element.get("environmentUuid") + self.key_name: str | None = raw_element.get("keyName") + + +class Notification(DynatraceObject): + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.key: str | None = raw_element.get("key") + self.account_uuid: str | None = raw_element.get("accountUuid") + self.message: str | None = raw_element.get("message") + self.severity: str | None = raw_element.get("severity") + self.type: str | None = raw_element.get("type") + details = raw_element.get("details") or {} + self.details: NotificationDetails | None = ( + NotificationDetails(raw_element=details) if details else None + ) + self.date: str | None = raw_element.get("date") + + +class NotificationList(DynatraceObject): + def _create_from_raw_data(self, raw_element: dict[str, Any]): + self.records: builtins.list[Notification] = [ + Notification(raw_element=record) + for record in raw_element.get("records", []) + ] + self.total_record_count: int | None = raw_element.get("totalRecordCount") + self.has_next_page: bool | None = raw_element.get("hasNextPage") diff --git a/dynatrace/account/schemas.py b/dynatrace/account/schemas.py index 9b92eef..c6382ea 100644 --- a/dynatrace/account/schemas.py +++ b/dynatrace/account/schemas.py @@ -1,5 +1,6 @@ """High-level account API aggregators.""" +from dynatrace.account.audits import AccountAuditsService 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 @@ -8,6 +9,7 @@ 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.notifications import NotificationService 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 @@ -17,6 +19,12 @@ class AccountAPI: def __init__(self, http_client: HttpClient) -> None: + # Account audit APIs. + self.audits: AccountAuditsService = AccountAuditsService(http_client) + + # Account notification APIs. + self.notifications: NotificationService = NotificationService(http_client) + # Account environment management APIs. self.env_v1: AccountEnvironmentsV1Service = AccountEnvironmentsV1Service( http_client diff --git a/pyproject.toml b/pyproject.toml index 37fa7c5..92c0221 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "dt-async" -version = "1.0.8" +version = "1.0.9" description = "Dynatrace async API Python client" readme = "README.md" authors = ["David Lopes "] diff --git a/test/account/test_audits.py b/test/account/test_audits.py new file mode 100644 index 0000000..238f39e --- /dev/null +++ b/test/account/test_audits.py @@ -0,0 +1,108 @@ +import json +from datetime import UTC, datetime +from unittest import mock + +from dynatrace import DynatraceAsync +from dynatrace.account.audits import AccountAuditsService, Audit, AuditsByAccount +from dynatrace.http_client import HttpClient + + +class MockResponse: + def __init__(self, json_data=None): + self._json_data = json_data + self.headers = {} + self.text = json.dumps(json_data) if json_data is not None else "" + self.status_code = 200 + + def json(self): + return self._json_data + + +async def test_audits_service_is_exposed(dt: DynatraceAsync): + assert isinstance(dt.account.audits, AccountAuditsService) + + +async def test_audits_list_returns_models(dt: DynatraceAsync): + account_uuid = "account-123" + start_time = datetime(2026, 1, 1, 0, 0, tzinfo=UTC) + end_time = datetime(2026, 1, 31, 23, 59, tzinfo=UTC) + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + expected_params = { + "startTime": "2026-01-01T00:00:00.000", + "endTime": "2026-01-31T23:59:00.000", + "addFields": "details,eventOutcome", + "filter": "eventType = 'CREATE'", + "limit": 25, + "scanLimitGigabyte": 10, + "resultSizeLimitMegabyte": 5, + } + if (method, path) == ("GET", f"/audit/v1/accounts/{account_uuid}"): + assert params == expected_params + return MockResponse( + { + "audits": [ + { + "eventId": "af1f98c9-c611-4056-841b-d039b1af3f98", + "timestamp": "2026-01-01T10:00:00Z", + "user": "user@example.com", + "resource": "POLICY", + "resourceName": "Standard User", + "eventProvider": "Identity & Account Management", + "eventType": "CREATE", + "accountUuid": account_uuid, + "authenticationClientId": "client-1", + "authenticationGrantType": "AUTHORIZATION_CODE", + "authenticationToken": "token-1", + "authenticationType": "OAUTH2", + "details": { + "json_before": '{"enabled": false}', + "json_after": '{"enabled": true}', + }, + "eventOutcome": "SUCCESS", + "eventReason": "Created from UI", + "eventVersion": "1.0.0", + "originAddress": "0.0.0.0", + "originSession": "session-1", + "originType": "REST", + "originXForwardedFor": "192.168.1.1", + "resourceId": "resource-1", + "environmentUuid": "environment-1", + "userOrganization": "CUSTOMER", + } + ], + "warnings": [{"message": "Your result has been limited to 1."}], + } + ) + + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + audits = await dt.account.audits.list( + account_uuid, + start_time=start_time, + end_time=end_time, + add_fields=["details", "eventOutcome"], + filter="eventType = 'CREATE'", + limit=25, + scan_limit_gigabyte=10, + result_size_limit_megabyte=5, + ) + + assert isinstance(audits, AuditsByAccount) + assert len(audits.audits) == 1 + assert isinstance(audits.audits[0], Audit) + assert audits.audits[0].resource_name == "Standard User" + assert audits.audits[0].details["json_after"] == '{"enabled": true}' + assert len(audits.warnings) == 1 + assert audits.warnings[0].message == "Your result has been limited to 1." diff --git a/test/account/test_notifications.py b/test/account/test_notifications.py new file mode 100644 index 0000000..97d79ec --- /dev/null +++ b/test/account/test_notifications.py @@ -0,0 +1,120 @@ +import json +from datetime import UTC, datetime +from unittest import mock + +from dynatrace import DynatraceAsync +from dynatrace.account.notifications import ( + Notification, + NotificationDetails, + NotificationList, + NotificationService, + NotificationSeverity, + NotificationSortField, + NotificationType, +) +from dynatrace.http_client import HttpClient + + +class MockResponse: + def __init__(self, json_data=None): + self._json_data = json_data + self.headers = {} + self.text = json.dumps(json_data) if json_data is not None else "" + self.status_code = 200 + + def json(self): + return self._json_data + + +async def test_notifications_service_is_exposed(dt: DynatraceAsync): + assert isinstance(dt.account.notifications, NotificationService) + + +async def test_notifications_list_returns_models(dt: DynatraceAsync): + account_uuid = "account-123" + start_date_time = datetime(2025, 12, 1, 10, 0, tzinfo=UTC) + end_date_time = datetime(2025, 12, 31, 23, 59, tzinfo=UTC) + + async def fake_make_request( + self, + path, + params=None, + headers=None, + method="GET", + data=None, + files=None, + query_params=None, + **kwargs, + ): + expected_body = { + "startDateTime": "2025-12-01T10:00:00.000", + "endDateTime": "2025-12-31T23:59:00.000", + "types": ["BUDGET", "BYOK_REVOKED"], + "severities": ["WARN", "INFO"], + "capabilities": ["LOG_MANAGEMENT_ANALYZE"], + "environments": ["abc12345"], + "page": 1, + "pageSize": 20, + "sorts": ["-date"], + } + if (method, path) == ("POST", f"/v1/accounts/{account_uuid}/notifications"): + assert params == expected_body + return MockResponse( + { + "records": [ + { + "key": "budget-key-example", + "accountUuid": account_uuid, + "message": "Message for budget 0 0", + "severity": "WARN", + "type": "budget", + "details": { + "environments": ["env-uuid"], + "capabilities": ["cap1"], + "allEnvironments": False, + "allCapabilities": True, + }, + "date": "2025-12-14T10:02:09.297Z", + }, + { + "key": "byok-key-example", + "accountUuid": account_uuid, + "message": "BYOK event message", + "severity": "WARN", + "type": "byok-revoked", + "details": { + "environmentUuid": "env-uuid", + "keyName": "key-name", + }, + "date": "2025-12-14T10:02:09.297Z", + }, + ], + "totalRecordCount": 2, + "hasNextPage": False, + } + ) + + raise AssertionError(f"Unexpected request: {method} {path}") + + with mock.patch.object(HttpClient, "make_request", new=fake_make_request): + notifications = await dt.account.notifications.list( + account_uuid, + start_date_time=start_date_time, + end_date_time=end_date_time, + types=[NotificationType.BUDGET, NotificationType.BYOK_REVOKED], + severities=[NotificationSeverity.WARN, NotificationSeverity.INFO], + capabilities=["LOG_MANAGEMENT_ANALYZE"], + environments=["abc12345"], + page=1, + page_size=20, + sorts=[NotificationSortField.DATE_DESC], + ) + + assert isinstance(notifications, NotificationList) + assert len(notifications.records) == 2 + assert isinstance(notifications.records[0], Notification) + assert isinstance(notifications.records[0].details, NotificationDetails) + assert notifications.records[0].details.capabilities == ["cap1"] + assert notifications.records[1].details.key_name == "key-name" + assert notifications.total_record_count == 2 + assert notifications.has_next_page is False