test: integration tests for akri and opcua checks

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

@@ -0,0 +1,5 @@
+# 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.
+# ----------------------------------------------------------------------------------------------

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

@@ -0,0 +1,156 @@
+# 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 typing import Any, Dict, Optional, Tuple
+from azure.cli.core.azclierror import CLIInternalError
+from azext_edge.edge.providers.edge_api.base import EdgeResourceApi
+from ....helpers import (
+    PLURAL_KEY,
+    find_extra_or_missing_names,
+    get_kubectl_workload_items,
+    run,
+    sort_kubectl_items_by_namespace
+)
+
+
+def assert_enumerate_resources(
+    post_deployment: dict,
+    description_name: str,
+    key_name: str,
+    resource_api: EdgeResourceApi,
+    resource_kinds: list,
+    present: bool = True
+):
+    key = f"enumerate{key_name}Api"
+    status = "success" if present else "skipped"
+    assert post_deployment[key]
+    assert post_deployment[key]["status"] == status
+    assert post_deployment[key]["description"] == f"Enumerate {description_name} API resources"
+
+    assert len(post_deployment[key]["targets"]) == 1
+    target_key = f"{resource_api.group}/{resource_api.version}"
+    assert target_key in post_deployment[key]["targets"]
+    evaluation = post_deployment[key]["targets"][target_key]["_all_"]
+    assert evaluation["conditions"] is None
+    assert evaluation["status"] == status
+    assert len(evaluation["evaluations"]) == 1
+    assert evaluation["evaluations"][0]["status"] == status
+    assert len(evaluation["evaluations"][0]["value"]) == len(resource_kinds)
+    for kind in evaluation["evaluations"][0]["value"]:
+        assert kind.lower() in resource_kinds
+
+
+# Used by Akri and OPCUA
+def assert_eval_core_service_runtime(
+    post_deployment: dict,
+    description_name: str,
+    pod_prefix: str,
+    resource_match: Optional[str] = None,
+):
+    assert post_deployment["evalCoreServiceRuntime"]
+    assert post_deployment["evalCoreServiceRuntime"]["description"] == f"Evaluate {description_name} core service"
+    overall_status = "success"
+    runtime_resource = post_deployment["evalCoreServiceRuntime"]["targets"]["coreServiceRuntimeResource"]
+    for namespace in runtime_resource.keys():
+        namespace_status = "success"
+        assert not runtime_resource[namespace]["conditions"]
+        evals = runtime_resource[namespace]["evaluations"]
+        kubectl_pods = get_kubectl_workload_items(
+            prefixes=pod_prefix,
+            service_type="pod",
+            resource_match=resource_match
+        )
+        results = [pod["value"]["name"] for pod in evals]
+        find_extra_or_missing_names(
+            resource_type="pods",
+            result_names=results,
+            expected_names=kubectl_pods.keys()
+        )
+
+        for pod in evals:
+            prefix, name = pod["name"].split("/")
+            assert prefix == "pod"
+            assert name == pod["value"]["name"]
+            assert pod["value"]["status.phase"] == kubectl_pods[name]["status"]["phase"]
+            expected_status = "success"
+            if pod["value"]["status.phase"] in ["Pending", "Unknown"]:
+                expected_status = "warning"
+                namespace_status = overall_status = expected_status
+            elif pod["value"]["status.phase"] == "Failed":
+                expected_status = "error"
+                if namespace_status == "success":
+                    namespace_status = expected_status
+                if overall_status == "success":
+                    overall_status = expected_status
+            assert pod["status"] == expected_status
+        assert runtime_resource[namespace]["status"] == namespace_status
+
+    assert post_deployment["evalCoreServiceRuntime"]["status"] == overall_status
+
+
+def assert_general_eval_custom_resources(
+    post_deployment: Dict[str, Any],
+    items: Dict[str, Any],
+    description_name: str,
+    resource_api: EdgeResourceApi,
+    resource_kind_present: bool,
+    include_all_namespace: bool = False
+):
+    # this should check general shared attributes for different services.
+    # specific target checks should be in separate functions
+    resource_plural = items[PLURAL_KEY]
+    key = None
+    for possible_key in post_deployment:
+        if possible_key.lower() == f"eval{resource_plural}":
+            key = possible_key
+            break
+
+    if not resource_kind_present:
+        assert key is None
+        return
+    assert post_deployment[key]
+    assert post_deployment[key]["description"].startswith(f"Evaluate {description_name}")
+    # for the ones that have spaces
+    assert post_deployment[key]["description"].replace(" ", "").endswith(resource_plural)
+
+    # check the target existence
+    sorted_items = sort_kubectl_items_by_namespace(items, include_all=include_all_namespace)
+    target_key = f"{resource_plural}.{resource_api.group}"
+    assert target_key in post_deployment[key]["targets"]
+    namespace_dict = post_deployment[key]["targets"][target_key]
+    for namespace, kubectl_items in sorted_items.items():
+        assert namespace in namespace_dict
+        check_names = []
+        for item in namespace_dict[namespace]["evaluations"]:
+            if item.get("name"):
+                check_names.append(item.get("names"))
+        # if using resource name filter, could have missing items
+        assert len(check_names) <= len(kubectl_items)
+        for name in check_names:
+            assert name in kubectl_items
+
+
+def run_check_command(
+    detail_level: str,
+    ops_service: str,
+    resource_api: EdgeResourceApi,
+    resource_kind: str,
+    resource_match: str,
+) -> Tuple[Dict[str, Any], bool]:
+    try:
+        aio_check = run(f"kubectl api-resources --api-group={resource_api.group}")
+        service_present = resource_api.group in aio_check
+    except CLIInternalError:
+        service_present = resource_api.is_deployed()
+    # note that the text decoder really does not like the emojis
+    command = f"az iot ops check --as-object --ops-service {ops_service} --detail-level {detail_level} "
+    if resource_kind:
+        command += f"--resources {resource_kind} "
+    if resource_match:
+        command += f"--resource-name {resource_match} "
+    result = run(command)
+
+    return {cond["name"]: cond for cond in result["postDeployment"]}, service_present

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

@@ -0,0 +1,108 @@
+# 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 typing import Any, Dict
+import pytest
+from knack.log import get_logger
+from azext_edge.edge.providers.check.common import ResourceOutputDetailLevel
+from azext_edge.edge.providers.edge_api import (
+    AkriResourceKinds, AKRI_API_V0
+)
+from .helpers import (
+    assert_enumerate_resources,
+    assert_eval_core_service_runtime,
+    assert_general_eval_custom_resources,
+    run_check_command
+)
+from ....helpers import get_kubectl_custom_items
+from ....generators import generate_names
+
+logger = get_logger(__name__)
+
+
+@pytest.mark.parametrize("detail_level", ResourceOutputDetailLevel.list())
+@pytest.mark.parametrize("resource_kind", AkriResourceKinds.list() + [None])
+# TODO: figure out if name match should be a general test vs each service (minimize test runs)
+@pytest.mark.parametrize("resource_match", [None, "*otel*", "akri-opcua*", generate_names()])
+def test_akri_check(init_setup, detail_level, resource_match, resource_kind):
+    post_deployment, akri_present = run_check_command(
+        detail_level=detail_level,
+        ops_service="akri",
+        resource_api=AKRI_API_V0,
+        resource_kind=resource_kind,
+        resource_match=resource_match
+    )
+
+    # overall api
+    assert_enumerate_resources(
+        post_deployment=post_deployment,
+        description_name="Akri",
+        key_name="Akri",
+        resource_api=AKRI_API_V0,
+        resource_kinds=AkriResourceKinds.list(),
+        present=akri_present,
+    )
+
+    if not resource_kind:
+        assert_eval_core_service_runtime(
+            post_deployment=post_deployment,
+            description_name="Akri",
+            pod_prefix="aio-akri-",
+            resource_match=resource_match,
+        )
+    else:
+        assert "evalCoreServiceRuntime" not in post_deployment
+
+    custom_resources = get_kubectl_custom_items(
+        resource_api=AKRI_API_V0,
+        resource_match=resource_match,
+        include_plural=True
+    )
+    assert_eval_configurations(
+        post_deployment=post_deployment,
+        custom_resources=custom_resources,
+        resource_kind=resource_kind
+    )
+    assert_eval_instances(
+        post_deployment=post_deployment,
+        custom_resources=custom_resources,
+        resource_kind=resource_kind
+    )
+
+
+def assert_eval_configurations(
+    post_deployment: Dict[str, Any],
+    custom_resources: Dict[str, Any],
+    resource_kind: str,
+):
+    resource_kind_present = resource_kind in [None, AkriResourceKinds.CONFIGURATION.value]
+    configurations = custom_resources[AkriResourceKinds.CONFIGURATION.value]
+    assert_general_eval_custom_resources(
+        post_deployment=post_deployment,
+        items=configurations,
+        description_name="Akri",
+        resource_api=AKRI_API_V0,
+        resource_kind_present=resource_kind_present
+    )
+    # TODO: add more as --as-object gets fixed, such as success conditions
+
+
+def assert_eval_instances(
+    post_deployment: Dict[str, Any],
+    custom_resources: Dict[str, Any],
+    resource_kind: str,
+):
+    resource_kind_present = resource_kind in [None, AkriResourceKinds.INSTANCE.value]
+    instances = custom_resources[AkriResourceKinds.INSTANCE.value]
+    assert_general_eval_custom_resources(
+        post_deployment=post_deployment,
+        items=instances,
+        description_name="Akri",
+        resource_api=AKRI_API_V0,
+        resource_kind_present=resource_kind_present,
+        include_all_namespace=True
+    )
+    # TODO: add more as --as-object gets fixed, such as success conditions

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

@@ -0,0 +1,85 @@
+# 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 typing import Any, Dict
+import pytest
+from knack.log import get_logger
+from azext_edge.edge.providers.check.common import ResourceOutputDetailLevel
+from azext_edge.edge.providers.edge_api import (
+    OpcuaResourceKinds, OPCUA_API_V1
+)
+from .helpers import (
+    assert_enumerate_resources,
+    assert_eval_core_service_runtime,
+    assert_general_eval_custom_resources,
+    run_check_command
+)
+from ....helpers import get_kubectl_custom_items
+from ....generators import generate_names
+
+logger = get_logger(__name__)
+
+
+@pytest.mark.parametrize("detail_level", ResourceOutputDetailLevel.list())
+@pytest.mark.parametrize("resource_kind", OpcuaResourceKinds.list() + [None])
+# TODO: figure out if name match should be a general test vs each service (minimize test runs)
+@pytest.mark.parametrize("resource_match", [None, "*opc-supervisor*", generate_names()])
+def test_opcua_check(init_setup, detail_level, resource_match, resource_kind):
+    post_deployment, opcua_present = run_check_command(
+        detail_level=detail_level,
+        ops_service="opcua",
+        resource_api=OPCUA_API_V1,
+        resource_kind=resource_kind,
+        resource_match=resource_match
+    )
+
+    # overall api
+    assert_enumerate_resources(
+        post_deployment=post_deployment,
+        description_name="OPC UA broker",
+        key_name="OpcUaBroker",
+        resource_api=OPCUA_API_V1,
+        resource_kinds=OpcuaResourceKinds.list(),
+        present=opcua_present,
+    )
+
+    if not resource_kind:
+        assert_eval_core_service_runtime(
+            post_deployment=post_deployment,
+            description_name="OPC UA broker",
+            pod_prefix=["aio-opc-", "opcplc-"],
+            resource_match=resource_match,
+        )
+    else:
+        assert "evalCoreServiceRuntime" not in post_deployment
+
+    custom_resources = get_kubectl_custom_items(
+        resource_api=OPCUA_API_V1,
+        resource_match=resource_match,
+        include_plural=True
+    )
+    assert_eval_asset_types(
+        post_deployment=post_deployment,
+        custom_resources=custom_resources,
+        resource_kind=resource_kind
+    )
+
+
+def assert_eval_asset_types(
+    post_deployment: dict,
+    custom_resources: Dict[str, Any],
+    resource_kind: str,
+):
+    asset_types = custom_resources[OpcuaResourceKinds.ASSET_TYPE.value]
+    resource_kind_present = resource_kind in [None, OpcuaResourceKinds.ASSET_TYPE.value]
+    assert_general_eval_custom_resources(
+        post_deployment=post_deployment,
+        items=asset_types,
+        description_name="OPC UA broker",
+        resource_api=OPCUA_API_V1,
+        resource_kind_present=resource_kind_present
+    )
+    # TODO: add more as --as-object gets fixed