test: Unit test anta.tools

anta/cli/utils.py

@@ -21,7 +21,7 @@
 
 logger = logging.getLogger(__name__)
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no_cover
     from anta.result_manager import ResultManager
 
 

anta/models.py

@@ -16,7 +16,7 @@
 from anta.result_manager.models import TestResult
 from anta.tools.misc import anta_log_exception, exc_to_str
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no_cover
     from anta.device import AntaDevice
 
 F = TypeVar("F", bound=Callable[..., Any])

anta/tools/misc.py

@@ -31,7 +31,10 @@ def anta_log_exception(exception: Exception, message: Optional[str] = None, call
     if __DEBUG__:
         calling_logger.exception(message)
     else:
-        calling_logger.error(f"{message} {exc_to_str(exception)}")
+        log_message = exc_to_str(exception)
+        if message is not None:
+            log_message = f"{message} {log_message}"
+        calling_logger.error(log_message)
 
 
 def exc_to_str(exception: Exception) -> str:

anta/tools/pydantic.py

@@ -7,7 +7,7 @@
 import logging
 from typing import TYPE_CHECKING, Any, Dict, List, Sequence
 
-if TYPE_CHECKING:
+if TYPE_CHECKING:  # pragma: no_cover
     from anta.result_manager.models import ListResult
 
 logger = logging.getLogger(__name__)

pyproject.toml

@@ -203,6 +203,9 @@ exclude_lines = [
 
     # Don't complain about abstract methods, they aren't run:
     "@(abc\\.)?abstractmethod",
+
+    # Don't complain about TYPE_CHECKING blocks
+    "if TYPE_CHECKING:",
 ]
 
 ignore_errors = true

tests/lib/fixture.py

@@ -2,13 +2,15 @@
 
 """Fixture for Anta Testing"""
 
+from typing import Callable
 from unittest.mock import MagicMock, create_autospec
 
 import pytest
 from aioeapi import Device
 from click.testing import CliRunner
 
 from anta.device import AntaDevice
+from anta.result_manager.models import ListResult, TestResult
 
 
 @pytest.fixture
@@ -31,6 +33,46 @@ def mocked_device(hw_model: str = "unknown_hw") -> MagicMock:
     return mock
 
 
+@pytest.fixture
+def test_result_factory(mocked_device: MagicMock) -> Callable[[int], TestResult]:
+    """
+    Return a anta.result_manager.models.TestResult object
+    """
+    # pylint: disable=redefined-outer-name
+
+    def _create(index: int = 0) -> TestResult:
+        """
+        Actual Factory
+        """
+        return TestResult(
+            name=mocked_device.name,
+            test=f"VerifyTest{index}",
+            test_category=["test"],
+            test_description=f"Verifies Test {index}",
+        )
+
+    return _create
+
+
+@pytest.fixture
+def list_result_factory(test_result_factory: Callable[[int], TestResult]) -> Callable[[int], ListResult]:
+    """
+    Return a ListResult with 'size' TestResult instanciated using the test_result_factory fixture
+    """
+    # pylint: disable=redefined-outer-name
+
+    def _factory(size: int = 0) -> ListResult:
+        """
+        Factory for ListResult entry of size entries
+        """
+        result = ListResult()
+        for i in range(size):
+            result.append(test_result_factory(i))
+        return result
+
+    return _factory
+
+
 @pytest.fixture
 def click_runner() -> CliRunner:
     """

tests/units/cli/test__init__.py

@@ -1,5 +1,5 @@
 """
-Tests for anta.cli.__init__.py
+Tests for anta.cli.__init__
 """
 
 from __future__ import annotations

tests/units/tools/test_get_value.py

@@ -0,0 +1,47 @@
+"""
+Tests for anta.tools.get_value
+"""
+
+from __future__ import annotations
+
+from contextlib import nullcontext as does_not_raise
+from typing import Any, Optional
+
+import pytest
+
+from anta.tools.get_value import get_value
+
+INPUT_DICT = {"test_value": 42, "nested_test": {"nested_value": 43}}
+
+
+@pytest.mark.parametrize(
+    "input_dict, key, default, required, org_key, separator, expected_result, expected_raise",
+    [
+        pytest.param({}, "test", None, False, None, None, None, does_not_raise(), id="empty dict"),
+        pytest.param(INPUT_DICT, "test_value", None, False, None, None, 42, does_not_raise(), id="simple key"),
+        pytest.param(INPUT_DICT, "nested_test.nested_value", None, False, None, None, 43, does_not_raise(), id="nested_key"),
+        pytest.param(INPUT_DICT, "missing_value", None, False, None, None, None, does_not_raise(), id="missing_value"),
+        pytest.param(INPUT_DICT, "missing_value_with_default", "default_value", False, None, None, "default_value", does_not_raise(), id="default"),
+        pytest.param(INPUT_DICT, "missing_required", None, True, None, None, None, pytest.raises(ValueError), id="required"),
+        pytest.param(INPUT_DICT, "missing_required", None, True, "custom_org_key", None, None, pytest.raises(ValueError), id="custom org_key"),
+        pytest.param(INPUT_DICT, "nested_test||nested_value", None, None, None, "||", 43, does_not_raise(), id="custom separator"),
+    ],
+)
+def test_get_value(
+    input_dict: dict[Any, Any],
+    key: str,
+    default: Optional[str],
+    required: bool,
+    org_key: Optional[str],
+    separator: Optional[str],
+    expected_result: str,
+    expected_raise: Any,
+) -> None:
+    """
+    Test get_value
+    """
+    # pylint: disable=too-many-arguments
+    kwargs = {"default": default, "required": required, "org_key": org_key, "separator": separator}
+    kwargs = {k: v for k, v in kwargs.items() if v is not None}
+    with expected_raise:
+        assert get_value(input_dict, key, **kwargs) == expected_result  # type: ignore

tests/units/tools/test_misc.py

@@ -0,0 +1,101 @@
+"""
+Tests for anta.tools.misc
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING, Optional
+from unittest.mock import patch
+
+import pytest
+
+from anta.tools.misc import anta_log_exception, exc_to_str, tb_to_str
+
+if TYPE_CHECKING:
+    from pytest import LogCaptureFixture
+
+
+def my_raising_function(exception: Exception) -> None:
+    """
+    dummy function to raise Exception
+    """
+    raise exception
+
+
+@pytest.mark.parametrize(
+    "exception, message, calling_logger, __DEBUG__value, expected_message",
+    [
+        pytest.param(ValueError("exception message"), None, None, False, "ValueError (exception message)", id="exception only"),
+        pytest.param(ValueError("exception message"), "custom message", None, False, "custom message ValueError (exception message)", id="custom message"),
+        pytest.param(
+            ValueError("exception message"),
+            "custom logger:",
+            logging.getLogger("custom"),
+            False,
+            "custom logger: ValueError (exception message)",
+            id="custom logger",
+        ),
+        pytest.param(ValueError("exception message"), "Use with custom message", None, True, "Use with custom message", id="__DEBUG__ on"),
+    ],
+)
+def test_anta_log_exception(
+    caplog: LogCaptureFixture,
+    exception: Exception,
+    message: Optional[str],
+    calling_logger: Optional[logging.Logger],
+    __DEBUG__value: bool,
+    expected_message: str,
+) -> None:
+    """
+    Test anta_log_exception
+    """
+
+    if calling_logger is not None:
+        # https://github.com/pytest-dev/pytest/issues/3697
+        calling_logger.propagate = True
+        caplog.set_level(logging.ERROR, logger=calling_logger.name)
+    else:
+        caplog.set_level(logging.ERROR)
+    print(caplog.__dict__)
+    # Need to raise to trigger nice stacktrace for __DEBUG__ == True
+    try:
+        my_raising_function(exception)
+    except ValueError as e:
+        with patch("anta.tools.misc.__DEBUG__", __DEBUG__value):
+            anta_log_exception(e, message=message, calling_logger=calling_logger)
+
+    # One log captured
+    assert len(caplog.record_tuples) == 1
+    logger, level, message = caplog.record_tuples[0]
+
+    if calling_logger is not None:
+        assert calling_logger.name == logger
+    else:
+        assert logger == "anta.tools.misc"
+
+    assert level == logging.ERROR
+    assert message == expected_message
+    # the only place where we can see the stracktrace is in the capture.text
+    if __DEBUG__value is True:
+        assert "Traceback" in caplog.text
+
+
+@pytest.mark.parametrize("exception, expected_output", [(ValueError("test"), "ValueError (test)"), (ValueError(), "ValueError")])
+def test_exc_to_str(exception: Exception, expected_output: str) -> None:
+    """
+    Test exc_to_str
+    """
+    assert exc_to_str(exception) == expected_output
+
+
+def test_tb_to_str() -> None:
+    """
+    Test tb_to_str
+    """
+    try:
+        my_raising_function(ValueError("test"))
+    except ValueError as e:
+        output = tb_to_str(e)
+        assert "Traceback" in output
+        assert 'my_raising_function(ValueError("test"))' in output

tests/units/tools/test_pydantic.py

@@ -0,0 +1,43 @@
+"""
+Tests for anta.tools.pydantic
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Callable
+
+import pytest
+
+from anta.tools.pydantic import pydantic_to_dict
+
+if TYPE_CHECKING:
+    from anta.result_manager.models import ListResult
+
+EXPECTED_ONE_ENTRY = [
+    {"name": "testdevice", "test": "VerifyTest0", "test_category": ["test"], "test_description": "Verifies Test 0", "result": "unset", "messages": []}
+]
+EXPECTED_THREE_ENTRIES = [
+    {"name": "testdevice", "test": "VerifyTest0", "test_category": ["test"], "test_description": "Verifies Test 0", "result": "unset", "messages": []},
+    {"name": "testdevice", "test": "VerifyTest1", "test_category": ["test"], "test_description": "Verifies Test 1", "result": "unset", "messages": []},
+    {"name": "testdevice", "test": "VerifyTest2", "test_category": ["test"], "test_description": "Verifies Test 2", "result": "unset", "messages": []},
+]
+
+
+@pytest.mark.parametrize(
+    "number_of_entries, expected",
+    [
+        pytest.param(0, [], id="empty"),
+        pytest.param(1, EXPECTED_ONE_ENTRY, id="one"),
+        pytest.param(3, EXPECTED_THREE_ENTRIES, id="three"),
+    ],
+)
+def test_pydantic_to_dict(
+    list_result_factory: Callable[[int], ListResult],
+    number_of_entries: int,
+    expected: dict[str, Any],
+) -> None:
+    """
+    Test pydantic_to_dict
+    """
+    list_result = list_result_factory(number_of_entries)
+    assert pydantic_to_dict(list_result) == expected