test: az iot ops check --pre updates

azext_edge/tests/edge/checks/int/helpers.py

@@ -4,7 +4,7 @@
 # Licensed under the MIT License. See License file in the project root for license information.
 # ----------------------------------------------------------------------------------------------
 
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 from azure.cli.core.azclierror import CLIInternalError
 from azext_edge.edge.providers.edge_api.base import EdgeResourceApi
 from ....helpers import (
@@ -133,6 +133,32 @@ def assert_general_eval_custom_resources(
             assert name in kubectl_items
 
 
+def combine_statuses(
+    status_list: List[str]
+):
+    final_status = "success"
+    for status in status_list:
+        if final_status == "success" and status in ["warning", "error", "skipped"]:
+            final_status = status
+        elif final_status in ["warning", "skipped"] and status == "error":
+            final_status = status
+    return final_status
+
+
+def expected_status(
+    success_or_fail: bool,
+    success_or_warning: Optional[bool] = None,
+    warning_or_fail: Optional[bool] = None
+):
+    status = "success" if success_or_fail else "error"
+    if any([
+        status == "success" and success_or_warning is False,
+        status == "error" and warning_or_fail is True
+    ]):
+        status = "warning"
+    return status
+
+
 def run_check_command(
     detail_level: str,
     ops_service: str,

azext_edge/tests/edge/checks/int/test_pre_post_int.py

@@ -6,8 +6,16 @@
 
 import pytest
 from knack.log import get_logger
+from kubernetes.utils import parse_quantity
 from azure.cli.core.azclierror import CLIInternalError
 from ....helpers import run
+from .helpers import combine_statuses, expected_status
+from azext_edge.edge.providers.check.common import (
+    AIO_SUPPORTED_ARCHITECTURES,
+    MIN_NODE_MEMORY,
+    MIN_NODE_STORAGE,
+    MIN_NODE_VCPU
+)
 
 logger = get_logger(__name__)
 
@@ -40,4 +48,88 @@ def test_check_pre_post(init_setup, post, pre):
     assert bool(result.get("preDeployment")) == expected_pre
     assert bool(result.get("postDeployment")) == expected_post
 
-    # TODO: add in pre deployment asserts
+    if not expected_pre:
+        # only check the pre results since the post results will be checked per service
+        return
+
+    assert len(result["preDeployment"]) == 2
+    k8s_result, node_result = result["preDeployment"]
+
+    # k8s evaluation
+    assert k8s_result["description"] == "Evaluate Kubernetes server"
+    assert k8s_result["name"] == "evalK8sVers"
+    assert "k8s" in k8s_result["targets"]
+    assert "_all_" in k8s_result["targets"]["k8s"]
+    k8s_target_result = k8s_result["targets"]["k8s"]["_all_"]
+    assert k8s_target_result["conditions"] == ["(k8s version)>=1.20"]
+    k8s_version = run("kubectl version -o json").get("serverVersion")
+
+    k8s_status = k8s_target_result["evaluations"][0]["status"]
+    if k8s_version:
+        expected_version = f"{k8s_version.get('major')}.{k8s_version.get('minor')}"
+        assert k8s_target_result["evaluations"][0]["value"] == expected_version
+        assert k8s_status == expected_status(expected_version >= "1.20")
+    assert k8s_result["status"] == k8s_target_result["status"] == k8s_status
+
+    # cluster node evaluation
+    assert node_result["description"] == "Evaluate cluster nodes"
+    assert node_result["name"] == "evalClusterNodes"
+    kubectl_nodes = run("kubectl get nodes -o json")["items"]
+
+    # num nodes
+    assert "cluster/nodes" in node_result["targets"]
+    assert "_all_" in node_result["targets"]["cluster/nodes"]
+    node_count_target = node_result["targets"]["cluster/nodes"]["_all_"]
+    assert node_count_target["conditions"] == ["len(cluster/nodes)>=1"]
+    assert node_count_target["evaluations"][0]["value"] == len(kubectl_nodes)
+    final_status = expected_status(
+        success_or_fail=len(kubectl_nodes) >= 1,
+        success_or_warning=len(kubectl_nodes) == 1
+    )
+    assert node_count_target["evaluations"][0]["status"] == node_count_target["status"] == final_status
+
+    # node eval
+    for node in kubectl_nodes:
+        node_name = node["metadata"]["name"]
+        node_key = f"cluster/nodes/{node_name}"
+        assert node_key in node_result["targets"]
+        assert "_all_" in node_result["targets"][node_key]
+        node_target = node_result["targets"][node_key]["_all_"]
+
+        assert node_target["conditions"] == [
+            f"info.architecture in ({','.join(AIO_SUPPORTED_ARCHITECTURES)})",
+            f"condition.cpu>={MIN_NODE_VCPU}",
+            f"condition.memory>={MIN_NODE_MEMORY}",
+            f"condition.ephemeral-storage>={MIN_NODE_STORAGE}"
+        ]
+
+        node_arch = node_target["evaluations"][0]["value"]["info.architecture"]
+        assert node_arch == node["status"]["nodeInfo"]["architecture"]
+        assert node_target["evaluations"][0]["status"] == expected_status(
+            node_arch in AIO_SUPPORTED_ARCHITECTURES
+        )
+
+        node_capacity = node["status"]["capacity"]
+        node_cpu = node_target["evaluations"][1]["value"]["condition.cpu"]
+        assert node_cpu == int(node_capacity["cpu"])
+        assert node_target["evaluations"][1]["status"] == expected_status(
+            node_cpu >= int(MIN_NODE_VCPU)
+        )
+
+        node_memory = node_target["evaluations"][2]["value"]["condition.memory"]
+        assert node_memory == int(parse_quantity(node_capacity["memory"]))
+        assert node_target["evaluations"][2]["status"] == expected_status(
+            node_memory >= parse_quantity(MIN_NODE_MEMORY)
+        )
+
+        node_storage = node_target["evaluations"][3]["value"]["condition.ephemeral-storage"]
+        assert node_storage == int(parse_quantity(node_capacity["ephemeral-storage"]))
+        assert node_target["evaluations"][3]["status"] == expected_status(
+            node_storage >= parse_quantity(MIN_NODE_STORAGE)
+        )
+
+        node_status = combine_statuses([cond["status"] for cond in node_target["evaluations"]])
+        assert node_status == node_target["status"]
+        final_status = combine_statuses([final_status, node_status])
+
+    assert final_status == node_result["status"]