azext_edge/edge/providers/check/base/display.py (131 lines of code) (raw):
# coding=utf-8
# ----------------------------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License file in the project root for license information.
# ----------------------------------------------------------------------------------------------
from knack.log import get_logger
from rich.console import Console, NewLine
from rich.padding import Padding
from typing import Any, Dict, List, Optional, Tuple
from .check_manager import CheckManager
from ..common import ALL_NAMESPACES_TARGET, COLOR_STR_FORMAT, DEFAULT_PADDING, DEFAULT_PROPERTY_DISPLAY_COLOR
from ....common import CheckTaskStatus
logger = get_logger(__name__)
def add_display_and_eval(
check_manager: CheckManager,
target_name: str,
display_text: str,
eval_status: str,
eval_value: str,
resource_name: Optional[str] = None,
namespace: str = ALL_NAMESPACES_TARGET,
padding: Tuple[int, int, int, int] = (0, 0, 0, 8)
) -> None:
check_manager.add_display(
target_name=target_name,
namespace=namespace,
display=Padding(display_text, padding)
)
check_manager.add_target_eval(
target_name=target_name,
namespace=namespace,
status=eval_status,
value=eval_value,
resource_name=resource_name
)
# TODO: test + refactor
def display_as_list(console: Console, result: Dict[str, Any]) -> None:
success_count: int = 0
warning_count: int = 0
error_count: int = 0
skipped_count: int = 0
def _increment_summary(status: str) -> None:
nonlocal success_count, warning_count, error_count, skipped_count
if not status:
return
if status == CheckTaskStatus.success.value:
success_count = success_count + 1
elif status == CheckTaskStatus.warning.value:
warning_count = warning_count + 1
elif status == CheckTaskStatus.error.value:
error_count = error_count + 1
elif status == CheckTaskStatus.skipped.value:
skipped_count = skipped_count + 1
def _print_summary() -> None:
from rich.panel import Panel
success_content = f"[green]{success_count} check(s) succeeded.[/green]"
warning_content = f"{warning_count} check(s) raised warnings."
warning_content = (
f"[green]{warning_content}[/green]" if not warning_count else f"[yellow]{warning_content}[/yellow]"
)
error_content = f"{error_count} check(s) raised errors."
error_content = f"[green]{error_content}[/green]" if not error_count else f"[red]{error_content}[/red]"
skipped_content = f"[bright_white]{skipped_count} check(s) were skipped[/bright_white]."
content = f"{success_content}\n{warning_content}\n{error_content}\n{skipped_content}"
console.print(Panel(content, title="Check Summary", expand=False))
def _enumerate_displays(checks: List[Dict[str, dict]]) -> None:
for check in checks:
status = check.get("status")
prefix_emoji = _get_emoji_from_status(status)
console.print(Padding(f"{prefix_emoji} {check['description']}", (0, 0, 0, 4)))
targets = check.get("targets", {})
for type in targets:
for namespace in targets[type]:
namespace_target = targets[type][namespace]
displays = namespace_target.get("displays", [])
status = namespace_target.get("status")
for (idx, disp) in enumerate(displays):
# display status indicator on each 'namespaced' grouping of displays
if all([idx == 0, status]):
prefix_emoji = _get_emoji_from_status(status)
console.print(Padding(f"\n{prefix_emoji} {disp.renderable}", (0, 0, 0, 6)))
else:
console.print(disp)
target_status = targets[type][namespace].get("status")
evaluations = targets[type][namespace].get("evaluations", [])
if not evaluations:
_increment_summary(target_status)
for e in evaluations:
eval_status = e.get("status")
_increment_summary(eval_status)
console.print(NewLine(1))
console.print(NewLine(1))
title: dict = result.get("title")
if title:
console.print(NewLine(1))
console.rule(title, align="center", style="blue bold")
console.print(NewLine(1))
pre_checks: List[dict] = result.get("preDeployment")
if pre_checks:
console.rule("Pre deployment checks", align="left")
console.print(NewLine(1))
_enumerate_displays(pre_checks)
post_checks: List[dict] = result.get("postDeployment")
if post_checks:
console.rule("Post deployment checks", align="left")
console.print(NewLine(1))
_enumerate_displays(post_checks)
_print_summary()
def _get_emoji_from_status(status: str) -> str:
return "" if not status else CheckTaskStatus.map_to_colored_emoji(status)
def process_value_color(
check_manager: CheckManager,
target_name: str,
key: Any,
value: Any,
) -> str:
value = value if value else "N/A"
if "error" in str(key).lower() and str(value).lower() not in ["null", "n/a", "none", "noerror"]:
check_manager.set_target_status(
target_name=target_name,
status=CheckTaskStatus.error.value
)
return f"[red]{value}[/red]"
return f"[cyan]{value}[/cyan]"
def colorize_string(value: str, color: Optional[str] = DEFAULT_PROPERTY_DISPLAY_COLOR) -> str:
color = color or DEFAULT_PROPERTY_DISPLAY_COLOR
return COLOR_STR_FORMAT.format(value=value, color=color)
def basic_property_display(
label: str,
value: str,
color: Optional[str] = DEFAULT_PROPERTY_DISPLAY_COLOR,
padding: Optional[int] = DEFAULT_PADDING
) -> Padding:
padding = padding or DEFAULT_PADDING
return Padding(
f"{label}: {colorize_string(value=value, color=color)}",
(0, 0, 0, padding)
)