Skip to content

Commit

Permalink
chore(black): Correct style for sanitycheck
Browse files Browse the repository at this point in the history
  • Loading branch information
lperdereau authored and lcaflc committed Dec 12, 2024
1 parent a95bd61 commit 27b3c73
Show file tree
Hide file tree
Showing 9 changed files with 259 additions and 254 deletions.
2 changes: 1 addition & 1 deletion src/pvecontrol/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ def _parser():

# sanitycheck parser
parser_sanitycheck = subparsers.add_parser("sanitycheck", help="Run Sanity checks on the cluster")
parser_sanitycheck.add_argument('--check', action='append', required=False, help="Check to run", default=[])
parser_sanitycheck.add_argument("--check", action="append", required=False, help="Check to run", default=[])
parser_sanitycheck.set_defaults(func=actions.cluster.action_sanitycheck)

# _test parser, hidden from help
Expand Down
18 changes: 9 additions & 9 deletions src/pvecontrol/cluster.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,21 +27,21 @@ def _initstatus(self):

self.storages = []
for storage in self.get_resources_storages():
self.storages.append(PVEStorage(storage.pop("node"), storage.pop("id"), storage.pop("shared"), **storage))
self.storages.append(PVEStorage(storage.pop("node"), storage.pop("id"), storage.pop("shared"), **storage))

self.nodes = []
for node in self._api.nodes.get():
self.nodes.append(PVENode(self._api, node["node"], node["status"], node))
self.nodes.append(PVENode(self._api, node["node"], node["status"], node))

self.tasks = []
for task in self._api.cluster.tasks.get():
logging.debug("Get task informations: %s"%(str(task)))
self.tasks.append(PVETask(self._api, task["upid"]))
logging.debug("Get task informations: %s" % (str(task)))
self.tasks.append(PVETask(self._api, task["upid"]))

self.ha = {
'groups': self._api.cluster.ha.groups.get(),
'manager_status': self._api.cluster.ha.status.manager_status.get(),
'resources': self._api.cluster.ha.resources.get()
"groups": self._api.cluster.ha.groups.get(),
"manager_status": self._api.cluster.ha.status.manager_status.get(),
"resources": self._api.cluster.ha.resources.get(),
}

def refresh(self):
Expand Down Expand Up @@ -82,8 +82,8 @@ def get_vm(self, vm_id):
result = None
node_name = None
for vm in self.get_resources_vms():
if vm['vmid'] == vm_id:
node_name = vm['node']
if vm["vmid"] == vm_id:
node_name = vm["node"]
break

for node in self.nodes:
Expand Down
171 changes: 89 additions & 82 deletions src/pvecontrol/sanitycheck/checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,108 +3,115 @@

from pvecontrol.utils import fonts, teminal_support_utf_8, terminal_support_colors


class CheckType(Enum):
HA = 'HIGH_AVAILABILITY'
Node = "NODE"
HA = "HIGH_AVAILABILITY"
Node = "NODE"


class CheckCode(Enum):
CRIT = 'CRITICAL'
WARN = 'WARNING'
INFO = 'INFO'
OK = 'OK'
CRIT = "CRITICAL"
WARN = "WARNING"
INFO = "INFO"
OK = "OK"


ICONS_UTF8 = {
CheckCode.CRIT.value: '❌',
CheckCode.WARN.value: '⚠️',
CheckCode.INFO.value: 'ℹ️',
CheckCode.OK.value: '✅',
CheckCode.CRIT.value: "❌",
CheckCode.WARN.value: "⚠️",
CheckCode.INFO.value: "ℹ️",
CheckCode.OK.value: "✅",
}

ICONS_ASCII = {
CheckCode.CRIT.value: '[CRIT]',
CheckCode.WARN.value: '[WARN]',
CheckCode.INFO.value: '[INFO]',
CheckCode.OK.value: '[OK]',
CheckCode.CRIT.value: "[CRIT]",
CheckCode.WARN.value: "[WARN]",
CheckCode.INFO.value: "[INFO]",
CheckCode.OK.value: "[OK]",
}

ICONS_COLORED_ASCII = {
CheckCode.CRIT.value: f'{fonts.RED}[CRIT]{fonts.END}',
CheckCode.WARN.value: f'{fonts.YELLOW}[WARN]{fonts.END}',
CheckCode.INFO.value: f'{fonts.BLUE}[INFO]{fonts.END}',
CheckCode.OK.value: f'{fonts.GREEN}[OK]{fonts.END}',
CheckCode.CRIT.value: f"{fonts.RED}[CRIT]{fonts.END}",
CheckCode.WARN.value: f"{fonts.YELLOW}[WARN]{fonts.END}",
CheckCode.INFO.value: f"{fonts.BLUE}[INFO]{fonts.END}",
CheckCode.OK.value: f"{fonts.GREEN}[OK]{fonts.END}",
}

def set_icons():
if teminal_support_utf_8():
return ICONS_UTF8
if terminal_support_colors():
return ICONS_COLORED_ASCII
return ICONS_ASCII

ICONS = set_icons()

class CheckMessage:
def __init__(self, code: CheckCode, message):
self.code = code
self.message = message

def display(self, padding_max_size):
padding = padding_max_size - len(self.message)
msg = f"{self.message}{padding * '.'}{ICONS[self.code.value]}"
print(msg)

def __len__(self):
return len(self.message)

class Check(ABC):

type = ""
name = ""
def set_icons():
if teminal_support_utf_8():
return ICONS_UTF8
if terminal_support_colors():
return ICONS_COLORED_ASCII
return ICONS_ASCII

def __init__(self, proxmox, messages = None):
if messages is None:
messages = []
self.proxmox = proxmox
self.messages = messages

@abstractmethod
def run(self):
pass
ICONS = set_icons()

@property
def status(self):
"""Define status by the most import status in messages"""
status = []
for msg in self.messages:
# exit early if most import code is found.
if CheckCode.CRIT == msg.code:
return CheckCode.CRIT
status.append(msg.code)

if CheckCode.WARN in status:
return CheckCode.WARN
class CheckMessage:
def __init__(self, code: CheckCode, message):
self.code = code
self.message = message

if CheckCode.INFO in status:
return CheckCode.INFO
def display(self, padding_max_size):
padding = padding_max_size - len(self.message)
msg = f"{self.message}{padding * '.'}{ICONS[self.code.value]}"
print(msg)

return CheckCode.OK
def __len__(self):
return len(self.message)

def add_messages(self, messages):
if isinstance(messages, CheckMessage):
self.messages.append(messages)
elif isinstance(messages, list):
self.messages += messages

def set_code(self, code: CheckCode):
self.code = code
class Check(ABC):

def display(self, padding_max_size):
if terminal_support_colors():
name = f"{fonts.BOLD}{self.name}{fonts.END}\n"
else:
name = f"{self.name}\n"
print(name)

for msg in self.messages:
msg.display(padding_max_size)
print()
type = ""
name = ""

def __init__(self, proxmox, messages=None):
if messages is None:
messages = []
self.proxmox = proxmox
self.messages = messages

@abstractmethod
def run(self):
pass

@property
def status(self):
"""Define status by the most import status in messages"""
status = []
for msg in self.messages:
# exit early if most import code is found.
if CheckCode.CRIT == msg.code:
return CheckCode.CRIT
status.append(msg.code)

if CheckCode.WARN in status:
return CheckCode.WARN

if CheckCode.INFO in status:
return CheckCode.INFO

return CheckCode.OK

def add_messages(self, messages):
if isinstance(messages, CheckMessage):
self.messages.append(messages)
elif isinstance(messages, list):
self.messages += messages

def set_code(self, code: CheckCode):
self.code = code

def display(self, padding_max_size):
if terminal_support_colors():
name = f"{fonts.BOLD}{self.name}{fonts.END}\n"
else:
name = f"{self.name}\n"
print(name)

for msg in self.messages:
msg.display(padding_max_size)
print()
72 changes: 36 additions & 36 deletions src/pvecontrol/sanitycheck/sanitychecks.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,52 +3,52 @@
from pvecontrol.sanitycheck.tests import DEFAULT_CHECKS, DEFAULT_CHECK_IDS


class SanityCheck():
class SanityCheck:

def __init__(self, proxmox: PVECluster):
self._proxmox = proxmox
self._checks = []
self._proxmox = proxmox
self._checks = []

def run(self, checks):
if not checks:
checks = DEFAULT_CHECK_IDS
if not checks:
checks = DEFAULT_CHECK_IDS

for check in checks:
if not check in DEFAULT_CHECK_IDS:
print(
f"Sanity check '{check}' doesn't exists.\n"
f"Here available values are:\n{', '.join(DEFAULT_CHECK_IDS)}"
)
return 1
for check in checks:
if not check in DEFAULT_CHECK_IDS:
print(
f"Sanity check '{check}' doesn't exists.\n"
f"Here available values are:\n{', '.join(DEFAULT_CHECK_IDS)}"
)
return 1

for id in checks:
check = DEFAULT_CHECKS[id](self._proxmox)
check.run()
self._checks.append(check)
for id in checks:
check = DEFAULT_CHECKS[id](self._proxmox)
check.run()
self._checks.append(check)

return self.get_exit_code()
return self.get_exit_code()

def get_exit_code(self):
for check in self._checks:
# exit early if most import code is found.
if CheckCode.CRIT == check.status:
return 1
return 0
for check in self._checks:
# exit early if most import code is found.
if CheckCode.CRIT == check.status:
return 1
return 0

def _get_longest_message(self):
size = 0
for check in self._checks:
for msg in check.messages:
if len(msg) > size:
size = len(msg)
return size + 1
size = 0
for check in self._checks:
for msg in check.messages:
if len(msg) > size:
size = len(msg)
return size + 1

def display(self):
size = self._get_longest_message()
current_type = None
for check in self._checks:
if current_type != check.type:
current_type = check.type
dash_size = int((size - (len(check.type.value) + 2))/2)
print(f"{dash_size*'-'} {check.type.value} {dash_size*'-'}\n")
check.display(size)
size = self._get_longest_message()
current_type = None
for check in self._checks:
if current_type != check.type:
current_type = check.type
dash_size = int((size - (len(check.type.value) + 2)) / 2)
print(f"{dash_size*'-'} {check.type.value} {dash_size*'-'}\n")
check.display(size)
7 changes: 2 additions & 5 deletions src/pvecontrol/sanitycheck/tests/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,7 @@
from .ha_groups import HaGroups
from .ha_vms import HaVms

DEFAULT_CHECKS = {
Nodes.id: Nodes,
HaGroups.id: HaGroups,
HaVms.id: HaVms
}

DEFAULT_CHECKS = {Nodes.id: Nodes, HaGroups.id: HaGroups, HaVms.id: HaVms}

DEFAULT_CHECK_IDS = DEFAULT_CHECKS.keys()
24 changes: 12 additions & 12 deletions src/pvecontrol/sanitycheck/tests/ha_groups.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,17 +3,17 @@

class HaGroups(Check):

id = "ha_groups"
type = CheckType.HA
name = "Check HA groups"
id = "ha_groups"
type = CheckType.HA
name = "Check HA groups"

def run(self):
for group in self.proxmox.ha['groups']:
num_nodes = len(group['nodes'].split(","))
if num_nodes < 2:
msg = f"Group {group['group']} contain only {num_nodes} node"
self.add_messages(CheckMessage(CheckCode.CRIT, msg))
def run(self):
for group in self.proxmox.ha["groups"]:
num_nodes = len(group["nodes"].split(","))
if num_nodes < 2:
msg = f"Group {group['group']} contain only {num_nodes} node"
self.add_messages(CheckMessage(CheckCode.CRIT, msg))

if not self.messages:
msg = "HA Group checked"
self.add_messages(CheckMessage(CheckCode.OK, msg))
if not self.messages:
msg = "HA Group checked"
self.add_messages(CheckMessage(CheckCode.OK, msg))
Loading

0 comments on commit 27b3c73

Please sign in to comment.