diff --git a/cmk/gui/rest_api_types/notifications_types.py b/cmk/gui/rest_api_types/notifications_types.py index 444c7791ea5..ebcb712ed96 100644 --- a/cmk/gui/rest_api_types/notifications_types.py +++ b/cmk/gui/rest_api_types/notifications_types.py @@ -5,14 +5,30 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import cast, ClassVar, Literal, Protocol +from typing import Any, cast, ClassVar, Literal, Protocol from cmk.utils.notify_types import ( - BuiltInPluginNames, + AsciiMailPluginModel, + CiscoPluginModel, CustomPluginName, - NotificationPluginNameStr, - NotifyPluginParams, + CustomPluginType, + IlertPluginModel, + is_known_plugin, + JiraIssuePluginModel, + MailPluginModel, + MicrosoftTeamsPluginModel, + MKEventdPluginModel, + NotifyPlugin, + OpsGenieIssuesPluginModel, + PagerDutyPluginModel, PluginOptions, + PushoverPluginModel, + ServiceNowPluginModel, + SignL4PluginModel, + SlackPluginModel, + SmsApiPluginModel, + SpectrumPluginModel, + SplunkPluginModel, ) from cmk.gui.rest_api_types.notifications_rule_types import ( @@ -60,25 +76,18 @@ WebhookURLOption, ) -PluginParamsOrNone = NotifyPluginParams | None -PluginMkFormatType = tuple[BuiltInPluginNames | CustomPluginName, PluginParamsOrNone] - -class NotificationPlugin(Protocol): +class PluginAdapter(Protocol): plugin_name: ClassVar @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams) -> NotificationPlugin: - raise NotImplementedError - - @classmethod - def from_api_request(cls, incoming: APINotifyPlugin) -> NotificationPlugin: + def from_api_request(cls, incoming: APINotifyPlugin) -> PluginAdapter: raise NotImplementedError def api_response(self) -> APINotifyPlugin: raise NotImplementedError - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: raise NotImplementedError @@ -109,8 +118,8 @@ class AsciiMailPlugin: ) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> AsciiMailPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: AsciiMailPluginModel | None) -> AsciiMailPlugin: + if pluginparams is None: return cls() return cls( @@ -193,7 +202,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -207,7 +216,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "bulk_sort_order": self.sort_order_for_bulk_notificaions.to_mk_file_format(), "disable_multiplexing": self.send_separate_notification_to_every_recipient.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(AsciiMailPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -255,8 +267,8 @@ class HTMLMailPlugin: ) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> HTMLMailPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: MailPluginModel | None) -> HTMLMailPlugin: + if pluginparams is None: return cls() return cls( @@ -369,7 +381,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -387,7 +399,11 @@ def to_mk_file_format(self) -> PluginMkFormatType: "no_floating_graphs": self.no_floating_graphs.to_mk_file_format(), "notifications_with_graphs": self.notifications_with_graphs.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + + return ( + self.__class__.plugin_name, + cast(MailPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -400,8 +416,8 @@ class CiscoWebexPlugin: disable_ssl_cert_verification: CheckboxTrueOrNone = field(default_factory=CheckboxTrueOrNone) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> CiscoWebexPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: CiscoPluginModel | None) -> CiscoWebexPlugin: + if pluginparams is None: return cls() return cls( @@ -451,7 +467,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -460,7 +476,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "ignore_ssl": self.disable_ssl_cert_verification.to_mk_file_format(), "proxy_url": self.http_proxy.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(CiscoPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -471,8 +490,8 @@ class MkEventDPlugin: ip_address_of_remote_ec: CheckboxWithStrValue = field(default_factory=CheckboxWithStrValue) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> MkEventDPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: MKEventdPluginModel | None) -> MkEventDPlugin: + if pluginparams is None: return cls() return cls( @@ -514,7 +533,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) @@ -522,7 +541,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "remote": self.ip_address_of_remote_ec.to_mk_file_format(), "facility": self.syslog_facility_to_use.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(MKEventdPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -538,8 +560,8 @@ class IlertPlugin: url_prefix_for_links_to_checkmk: CheckboxURLPrefix = field(default_factory=CheckboxURLPrefix) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> IlertPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: IlertPluginModel | None) -> IlertPlugin: + if pluginparams is None: return cls() return cls( @@ -596,7 +618,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -608,7 +630,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "url_prefix": self.url_prefix_for_links_to_checkmk.to_mk_file_format(), "proxy_url": self.http_proxy.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(IlertPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -633,13 +658,10 @@ class JiraIssuePlugin: timeout: CheckboxWithStrValue = field(default_factory=CheckboxWithStrValue) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> JiraIssuePlugin: + def from_mk_file_format(cls, pluginparams: JiraIssuePluginModel | None) -> JiraIssuePlugin: if pluginparams is None: return cls() - if isinstance(pluginparams, list): - return cls() - return cls( option=PluginOptions.WITH_PARAMS, url=pluginparams["url"], @@ -733,7 +755,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -754,7 +776,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "resolution": self.resolution.to_mk_file_format(), "timeout": self.timeout.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(JiraIssuePluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -779,8 +804,10 @@ class OpsGenieIssuePlugin: entity: CheckboxWithStrValue = field(default_factory=CheckboxWithStrValue) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> OpsGenieIssuePlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format( + cls, pluginparams: OpsGenieIssuesPluginModel | None + ) -> OpsGenieIssuePlugin: + if pluginparams is None: return cls() return cls( @@ -888,7 +915,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -910,7 +937,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "entity": self.entity.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(OpsGenieIssuesPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -926,8 +956,8 @@ class PagerDutyPlugin: ) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> PagerDutyPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: PagerDutyPluginModel | None) -> PagerDutyPlugin: + if pluginparams is None: return cls() return cls( @@ -978,7 +1008,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -988,7 +1018,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "url_prefix": self.url_prefix_for_links_to_checkmk.to_mk_file_format(), "webhook_url": self.webhook_url, } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(PagerDutyPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1005,8 +1038,8 @@ class PushOverPlugin: sound: CheckboxPushoverSound = field(default_factory=CheckboxPushoverSound) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> PushOverPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: PushoverPluginModel | None) -> PushOverPlugin: + if pluginparams is None: return cls() return cls( @@ -1062,7 +1095,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1073,7 +1106,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "sound": self.sound.to_mk_file_format(), "url_prefix": self.url_prefix_for_links_to_checkmk.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(PushoverPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1089,8 +1125,8 @@ class ServiceNowPlugin: mgmt_type: ManagementType = field(default_factory=ManagementType) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> ServiceNowPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: ServiceNowPluginModel | None) -> ServiceNowPlugin: + if pluginparams is None: return cls() return cls( @@ -1145,7 +1181,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1157,7 +1193,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "timeout": self.timeout.to_mk_file_format(), "mgmt_type": self.mgmt_type.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(ServiceNowPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1170,13 +1209,10 @@ class SignL4Plugin: http_proxy: CheckboxHttpProxy = field(default_factory=CheckboxHttpProxy) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> SignL4Plugin: + def from_mk_file_format(cls, pluginparams: SignL4PluginModel | None) -> SignL4Plugin: if pluginparams is None: return cls() - if isinstance(pluginparams, list): - return cls() - return cls( option=PluginOptions.WITH_PARAMS, team_secret=APISignL4SecretOption.from_mk_file_format(pluginparams["password"]), @@ -1224,7 +1260,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1233,7 +1269,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "ignore_ssl": self.disable_ssl_cert_verification.to_mk_file_format(), "proxy_url": self.http_proxy.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(SignL4PluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1246,8 +1285,8 @@ class SlackPlugin: http_proxy: CheckboxHttpProxy = field(default_factory=CheckboxHttpProxy) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> SlackPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: SlackPluginModel | None) -> SlackPlugin: + if pluginparams is None: return cls() return cls( @@ -1299,7 +1338,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1308,7 +1347,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "ignore_ssl": self.disable_ssl_cert_verification.to_mk_file_format(), "proxy_url": self.http_proxy.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(SlackPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1324,8 +1366,8 @@ class SMSAPIPlugin: timeout: str | None = None @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> SMSAPIPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: SmsApiPluginModel | None) -> SMSAPIPlugin: + if pluginparams is None: return cls() return cls( @@ -1378,7 +1420,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1390,7 +1432,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "password": self.user_password.to_mk_file_format(), "timeout": self.timeout, } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(SmsApiPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1400,8 +1445,8 @@ class SMSPlugin: params: list[str] | None = None @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> SMSPlugin: - if pluginparams is None or isinstance(pluginparams, dict): + def from_mk_file_format(cls, pluginparams: list[str] | None) -> SMSPlugin: + if pluginparams is None: return cls() return cls( @@ -1430,7 +1475,7 @@ def api_response(self) -> APINotifyPlugin: def to_mk_file_format( self, - ) -> PluginMkFormatType: + ) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) return (self.__class__.plugin_name, self.params) @@ -1445,8 +1490,8 @@ class SpectrumPlugin: destination_ip: str = "" @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> SpectrumPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: SpectrumPluginModel | None) -> SpectrumPlugin: + if pluginparams is None: return cls() return cls( @@ -1483,7 +1528,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1491,7 +1536,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "community": self.snmp_community, "destination": self.destination_ip, } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(SpectrumPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1504,8 +1552,8 @@ class VictoropsPlugin: splunk_on_call_rest_endpoint: WebhookURLOption = field(default_factory=WebhookURLOption) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> VictoropsPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: SplunkPluginModel | None) -> VictoropsPlugin: + if pluginparams is None: return cls() return cls( @@ -1560,7 +1608,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1569,7 +1617,10 @@ def to_mk_file_format(self) -> PluginMkFormatType: "url_prefix": self.url_prefix_for_links_to_checkmk.to_mk_file_format(), "webhook_url": self.splunk_on_call_rest_endpoint.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(SplunkPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass @@ -1604,8 +1655,8 @@ class MsTeamsPlugin: ) @classmethod - def from_mk_file_format(cls, pluginparams: NotifyPluginParams | None) -> MsTeamsPlugin: - if pluginparams is None or isinstance(pluginparams, list): + def from_mk_file_format(cls, pluginparams: MicrosoftTeamsPluginModel | None) -> MsTeamsPlugin: + if pluginparams is None: return cls() return cls( @@ -1688,7 +1739,7 @@ def api_response(self) -> APINotifyPlugin: r: APINotifyPlugin = {"option": self.option, "plugin_params": params} return r - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: if self.option == "cancel_previous_notifications": return (self.__class__.plugin_name, None) r = { @@ -1703,29 +1754,32 @@ def to_mk_file_format(self) -> PluginMkFormatType: "url_prefix": self.url_prefix_for_links_to_checkmk.to_mk_file_format(), "webhook_url": self.webhook_url.to_mk_file_format(), } - return (self.__class__.plugin_name, {k: v for k, v in r.items() if v is not None}) + return ( + self.__class__.plugin_name, + cast(MicrosoftTeamsPluginModel, {k: v for k, v in r.items() if v is not None}), + ) @dataclass -class CustomPlugin: +class CustomPluginAdapter: plugin_name: CustomPluginName option: PluginOptions = PluginOptions.CANCEL - plugin_options: NotifyPluginParams | None = None + plugin_options: list[str] | dict[str, Any] | None = None @classmethod def from_mk_file_format( - cls, plugin_name: CustomPluginName, pluginparams: NotifyPluginParams | None - ) -> CustomPlugin: + cls, plugin_name: str, pluginparams: list[str] | dict[str, Any] | None + ) -> CustomPluginAdapter: if pluginparams is None: - return cls(plugin_name=plugin_name) + return cls(plugin_name=CustomPluginName(plugin_name)) return cls( - plugin_name=plugin_name, + plugin_name=CustomPluginName(plugin_name), option=PluginOptions.WITH_CUSTOM_PARAMS, plugin_options=pluginparams, ) @classmethod - def from_api_request(cls, incoming: APINotifyPlugin) -> CustomPlugin: + def from_api_request(cls, incoming: APINotifyPlugin) -> CustomPluginAdapter: option = incoming["option"] plugin_params = incoming["plugin_params"] plugin_name = cast(CustomPluginName, plugin_params["plugin_name"]) @@ -1768,57 +1822,72 @@ def api_response(self) -> APINotifyPlugin: "plugin_params": plugin_params, } - def to_mk_file_format(self) -> PluginMkFormatType: + def to_mk_file_format(self) -> NotifyPlugin: return self.plugin_name, self.plugin_options -def get_plugin_from_mk_file( - plugin_name: NotificationPluginNameStr | CustomPluginName, - notifypluginparams: NotifyPluginParams | None, -) -> NotificationPlugin | CustomPlugin: - match plugin_name: - case "cisco_webex_teams": - return CiscoWebexPlugin.from_mk_file_format(notifypluginparams) - case "mkeventd": - return MkEventDPlugin.from_mk_file_format(notifypluginparams) - case "asciimail": - return AsciiMailPlugin.from_mk_file_format(notifypluginparams) - case "mail": - return HTMLMailPlugin.from_mk_file_format(notifypluginparams) - case "msteams": - return MsTeamsPlugin.from_mk_file_format(notifypluginparams) - case "ilert": - return IlertPlugin.from_mk_file_format(notifypluginparams) - case "jira_issues": - return JiraIssuePlugin.from_mk_file_format(notifypluginparams) - case "opsgenie_issues": - return OpsGenieIssuePlugin.from_mk_file_format(notifypluginparams) - case "pagerduty": - return PagerDutyPlugin.from_mk_file_format(notifypluginparams) - case "pushover": - return PushOverPlugin.from_mk_file_format(notifypluginparams) - case "servicenow": - return ServiceNowPlugin.from_mk_file_format(notifypluginparams) - case "signl4": - return SignL4Plugin.from_mk_file_format(notifypluginparams) - case "slack": - return SlackPlugin.from_mk_file_format(notifypluginparams) - case "sms_api": - return SMSAPIPlugin.from_mk_file_format(notifypluginparams) - case "sms": - return SMSPlugin.from_mk_file_format(notifypluginparams) - case "spectrum": - return SpectrumPlugin.from_mk_file_format(notifypluginparams) - case "victorops": - return VictoropsPlugin.from_mk_file_format(notifypluginparams) - case _: - return CustomPlugin.from_mk_file_format( - plugin_name, - notifypluginparams, - ) +def get_plugin_from_mk_file( # pylint: disable=too-many-branches + notify_plugin: NotifyPlugin, +) -> PluginAdapter | CustomPluginAdapter: + + # TODO use match case once mypy has support for it + if is_known_plugin(notify_plugin): + if notify_plugin[0] == "sms": + return SMSPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "cisco_webex_teams": + return CiscoWebexPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "mkeventd": + return MkEventDPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "asciimail": + return AsciiMailPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "mail": + return HTMLMailPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "msteams": + return MsTeamsPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "ilert": + return IlertPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "jira_issues": + return JiraIssuePlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "opsgenie_issues": + return OpsGenieIssuePlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "pagerduty": + return PagerDutyPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "pushover": + return PushOverPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "servicenow": + return ServiceNowPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "signl4": + return SignL4Plugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "slack": + return SlackPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "sms_api": + return SMSAPIPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "spectrum": + return SpectrumPlugin.from_mk_file_format(notify_plugin[1]) + + if notify_plugin[0] == "victorops": + return VictoropsPlugin.from_mk_file_format(notify_plugin[1]) + + custom_plugin = cast(CustomPluginType, notify_plugin) + return CustomPluginAdapter.from_mk_file_format(custom_plugin[0], custom_plugin[1]) -def get_plugin_from_api_request(incoming: APINotifyPlugin) -> NotificationPlugin | CustomPlugin: +def get_plugin_from_api_request(incoming: APINotifyPlugin) -> PluginAdapter | CustomPluginAdapter: match incoming["plugin_params"]["plugin_name"]: case "cisco_webex_teams": return CiscoWebexPlugin.from_api_request(incoming) @@ -1855,4 +1924,4 @@ def get_plugin_from_api_request(incoming: APINotifyPlugin) -> NotificationPlugin case "victorops": return VictoropsPlugin.from_api_request(incoming) case _: - return CustomPlugin.from_api_request(incoming) + return CustomPluginAdapter.from_api_request(incoming) diff --git a/cmk/gui/watolib/notifications.py b/cmk/gui/watolib/notifications.py index 9a63a660689..459ee201221 100644 --- a/cmk/gui/watolib/notifications.py +++ b/cmk/gui/watolib/notifications.py @@ -30,13 +30,7 @@ from typing import Any, cast import cmk.utils.store as store -from cmk.utils.notify_types import ( - BuiltInPluginNames, - EventRule, - NotificationRuleID, - NotifyBulkType, - NotifyPlugin, -) +from cmk.utils.notify_types import EventRule, NotificationRuleID, NotifyBulkType, NotifyPlugin from cmk.utils.user import UserId import cmk.gui.userdb as userdb @@ -63,10 +57,10 @@ RestrictToNotificationNumbers, ) from cmk.gui.rest_api_types.notifications_types import ( - CustomPlugin, + CustomPluginAdapter, get_plugin_from_api_request, get_plugin_from_mk_file, - NotificationPlugin, + PluginAdapter, ) from cmk.gui.type_defs import GlobalSettings from cmk.gui.watolib.simple_config_file import ConfigFileRegistry, WatoListConfigFile @@ -197,17 +191,14 @@ class BulkNotAllowedException(Exception): ... @dataclass class NotificationMethod: notification_bulking: CheckboxNotificationBulking - notify_plugin: NotificationPlugin | CustomPlugin + notify_plugin: PluginAdapter | CustomPluginAdapter @classmethod def from_mk_file_format( cls, notify_plugin: NotifyPlugin, bulk_config: NotifyBulkType | None ) -> NotificationMethod: - plugin_name, pluginparams = notify_plugin - builtin_plugin_name = cast(BuiltInPluginNames, plugin_name) - return cls( - notify_plugin=get_plugin_from_mk_file(builtin_plugin_name, pluginparams), + notify_plugin=get_plugin_from_mk_file(notify_plugin), notification_bulking=CheckboxNotificationBulking.from_mk_file_format(bulk_config), ) diff --git a/cmk/gui/watolib/sample_config.py b/cmk/gui/watolib/sample_config.py index 988ce61ac31..29ef3062bfb 100644 --- a/cmk/gui/watolib/sample_config.py +++ b/cmk/gui/watolib/sample_config.py @@ -12,7 +12,7 @@ from cmk.utils import store from cmk.utils.encryption import raw_certificates_from_file from cmk.utils.log import VERBOSE -from cmk.utils.notify_types import EventRule, NotificationRuleID +from cmk.utils.notify_types import EventRule, MailPluginModel, NotificationRuleID from cmk.utils.paths import site_cert_file from cmk.utils.tags import sample_tag_config, TagConfig @@ -103,7 +103,7 @@ def get_default_notification_rule() -> EventRule: contact_object=True, description="Notify all contacts of a host/service via HTML email", disabled=False, - notify_plugin=("mail", {}), + notify_plugin=("mail", MailPluginModel()), ) diff --git a/cmk/utils/notify_types.py b/cmk/utils/notify_types.py index 5811bbebb5f..cfdb881e517 100644 --- a/cmk/utils/notify_types.py +++ b/cmk/utils/notify_types.py @@ -5,7 +5,19 @@ from collections.abc import Mapping, Sequence from enum import StrEnum -from typing import Any, Literal, NewType, NotRequired, Required, TypedDict, TypeGuard +from typing import ( + Annotated, + Any, + get_args, + Literal, + NewType, + NotRequired, + Required, + TypedDict, + TypeGuard, +) + +from pydantic import PlainValidator, TypeAdapter, ValidationInfo from cmk.utils.hostaddress import HostName from cmk.utils.timeperiod import TimeperiodName @@ -71,9 +83,6 @@ class TimeperiodBulkParameters(BulkBaseParameters): bulk_outside: NotRequired[AlwaysBulkParameters] -NotifyPluginParamsList = list[str] -NotifyPluginParamsDict = dict[str, Any] # TODO: Improve this -NotifyPluginParams = NotifyPluginParamsList | NotifyPluginParamsDict NotifyBulkParameters = AlwaysBulkParameters | TimeperiodBulkParameters NotifyBulkType = ( tuple[Literal["always"], AlwaysBulkParameters] @@ -129,7 +138,7 @@ class PluginOptions(StrEnum): ] CustomPluginName = NewType("CustomPluginName", str) -NotificationPluginNameStr = BuiltInPluginNames +NotificationPluginNameStr = BuiltInPluginNames | CustomPluginName MgmntPriorityType = Literal[ "low", @@ -442,28 +451,9 @@ class BulkParameters(TypedDict, total=False): PluginNotificationContext = dict[str, str] -NotifyPlugin = tuple[NotificationPluginNameStr, NotifyPluginParams | None] NotificationRuleID = NewType("NotificationRuleID", str) -MailElement = Literal[ - "omdsite", - "hosttags", - "address", - "abstime", - "reltime", - "longoutput", - "ack_author", - "ack_comment", - "notification_author", - "notification_comment", - "perfdata", - "graph", - "notesurl", - "context", -] - - class EmailFromOrTo(TypedDict): display_name: NotRequired[str] address: NotRequired[str] @@ -478,7 +468,7 @@ class EmailFromOrTo(TypedDict): "service_subject": NotRequired[str], "bulk_sort_order": NotRequired[Literal["oldest_first", "newest_first"]], "disable_multiplexing": NotRequired[Literal[True]], - "elements": NotRequired[list[MailElement]], + "elements": NotRequired[list[EmailBodyElementsType]], "insert_html_section": NotRequired[str], "url_prefix": NotRequired[URLPrefix], "no_floating_graphs": NotRequired[Literal[True]], @@ -713,6 +703,75 @@ class SplunkPluginModel(TypedDict, total=False): SpectrumNotify = tuple[Literal["spectrum"], SpectrumPluginModel | None] SplunkNotify = tuple[Literal["victorops"], SplunkPluginModel | None] +KnownPlugins = ( + MailNotify + | AsciiMailNotify + | CiscoNotify + | MkeventdNotify + | IlertNotify + | JiraNotify + | OpsgenieNotify + | PagerdutyNotify + | PushoverNotify + | ServiceNowNotify + | SignL4Notify + | SlackNotify + | SmsApiNotify + | SmsNotify + | SpectrumNotify + | SplunkNotify + | MSteamsNotify +) + +CustomPluginType = tuple[CustomPluginName, dict[str, Any] | list[str] | None] + +NotifyPlugin = KnownPlugins | CustomPluginType + + +def is_known_plugin(notify_plugin: NotifyPlugin) -> TypeGuard[KnownPlugins]: + return notify_plugin[0] in get_args(BuiltInPluginNames) + + +NotifyPluginParamsList = list[str] +NotifyPluginParamsDict = ( + MailPluginModel + | AsciiMailPluginModel + | CiscoPluginModel + | MKEventdPluginModel + | IlertPluginModel + | JiraIssuePluginModel + | OpsGenieIssuesPluginModel + | PagerDutyPluginModel + | PushoverPluginModel + | ServiceNowPluginModel + | SignL4PluginModel + | SlackPluginModel + | SmsApiPluginModel + | SpectrumPluginModel + | SplunkPluginModel + | MicrosoftTeamsPluginModel + | dict[str, Any] +) + +NotifyPluginParams = NotifyPluginParamsList | NotifyPluginParamsDict + + +custom_plugin_type_adapter = TypeAdapter(CustomPluginType) +known_plugin_type_adapter = TypeAdapter(KnownPlugins) + + +def validate_plugin(value: Any, _handler: ValidationInfo) -> NotifyPlugin: + assert isinstance(value, tuple) + assert len(value) == 2 + + # If it's a builtin plugin, validate against it's corresponding typeddict. + if value[0] in get_args(BuiltInPluginNames): + known_plugin_type_adapter.validate_python(value, strict=True) + return value + + custom_plugin_type_adapter.validate_python(value) + return value + class _EventRuleMandatory(TypedDict): rule_id: NotificationRuleID @@ -722,7 +781,7 @@ class _EventRuleMandatory(TypedDict): contact_object: bool description: str disabled: bool - notify_plugin: NotifyPlugin + notify_plugin: Annotated[NotifyPlugin, PlainValidator(validate_plugin)] class EventRule(_EventRuleMandatory, total=False): @@ -767,11 +826,9 @@ class EventRule(_EventRuleMandatory, total=False): match_site: list[str] match_sl: tuple[int, int] match_timeperiod: TimeperiodName - notify_method: NotifyPluginParams bulk: NotifyBulkType match_service_level: tuple[int, int] match_only_during_timeperiod: str - notification_method: NotificationPluginNameStr NotifyRuleInfo = tuple[str, EventRule, str]