From c24a8fba3366246987da95a21d80dbb681070d97 Mon Sep 17 00:00:00 2001 From: Konstantin Baikov Date: Thu, 16 May 2024 10:12:38 +0200 Subject: [PATCH] EC: add micro type fixes found by pyright Change-Id: I7fa31a3e206d469606283aa3c9c64e3d6dc8094f --- cmk/ec/main.py | 2 +- cmk/ec/rule_packs.py | 17 ++++++++++------- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/cmk/ec/main.py b/cmk/ec/main.py index a5d08c92f8f..e74782fd551 100644 --- a/cmk/ec/main.py +++ b/cmk/ec/main.py @@ -3219,6 +3219,7 @@ def get_state_from_master(config: Config, slave_status: SlaveStatus) -> Any: repl_settings = config["replication"] if repl_settings is None: raise ValueError("no replication settings") + response_text = b"" try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(repl_settings["connect_timeout"]) @@ -3228,7 +3229,6 @@ def get_state_from_master(config: Config, slave_status: SlaveStatus) -> Any: ) sock.shutdown(socket.SHUT_WR) - response_text = b"" while True: chunk = sock.recv(8192) response_text += chunk diff --git a/cmk/ec/rule_packs.py b/cmk/ec/rule_packs.py index 5b0182e0d2e..c56d80bafab 100644 --- a/cmk/ec/rule_packs.py +++ b/cmk/ec/rule_packs.py @@ -24,10 +24,13 @@ from .config import ( ConfigFromWATO, + ContactGroups, ECRulePack, ECRulePackSpec, + LogConfig, MkpRulePackBindingError, MkpRulePackProxy, + ServiceLevel, ) from .defaults import default_config, default_rule_pack from .settings import create_paths, Settings @@ -134,17 +137,17 @@ def _load_config( # pylint: disable=too-many-branches for rule in rule_pack["rules"]: # Convert old contact_groups config if isinstance(rule.get("contact_groups"), list): - rule["contact_groups"] = { - "groups": rule["contact_groups"], - "notify": False, - "precedence": "host", - } + rule["contact_groups"] = ContactGroups( + groups=rule["contact_groups"], + notify=False, + precedence="host", + ) # Old configs only have a naked service level without a precedence. if isinstance(rule["sl"], int): - rule["sl"] = {"value": rule["sl"], "precedence": "message"} + rule["sl"] = ServiceLevel(value=rule["sl"], precedence="message") # Convert old logging configurations - levels = config["log_level"] + levels: LogConfig = config["log_level"] if isinstance(levels, int): # TODO: Move this to upgrade time level = logging.INFO if levels == 0 else cmk.utils.log.VERBOSE # type: ignore[unreachable] levels = {