summaryrefslogtreecommitdiffstats
path: root/tests/units/result_manager
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2025-01-14 10:18:19 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2025-01-14 10:18:29 +0000
commit6818d016122ee845a2011b94bbdad0ed28a9aae7 (patch)
treee9865932680acf05b8c353347cf362ab3fd10ff0 /tests/units/result_manager
parentReleasing debian version 1.1.0-1. (diff)
downloadanta-6818d016122ee845a2011b94bbdad0ed28a9aae7.tar.xz
anta-6818d016122ee845a2011b94bbdad0ed28a9aae7.zip
Merging upstream version 1.2.0.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'tests/units/result_manager')
-rw-r--r--tests/units/result_manager/test__init__.py101
1 files changed, 101 insertions, 0 deletions
diff --git a/tests/units/result_manager/test__init__.py b/tests/units/result_manager/test__init__.py
index 1fd51cb..e41a436 100644
--- a/tests/units/result_manager/test__init__.py
+++ b/tests/units/result_manager/test__init__.py
@@ -6,6 +6,7 @@
from __future__ import annotations
import json
+import logging
import re
from contextlib import AbstractContextManager, nullcontext
from typing import TYPE_CHECKING, Callable
@@ -379,3 +380,103 @@ class TestResultManager:
assert len(result_manager.get_devices()) == 2
assert all(t in result_manager.get_devices() for t in ["Device1", "Device2"])
+
+ def test_stats_computation_methods(self, test_result_factory: Callable[[], TestResult], caplog: pytest.LogCaptureFixture) -> None:
+ """Test ResultManager internal stats computation methods."""
+ result_manager = ResultManager()
+
+ # Initially stats should be unsynced
+ assert result_manager._stats_in_sync is False
+
+ # Test _reset_stats
+ result_manager._reset_stats()
+ assert result_manager._stats_in_sync is False
+ assert len(result_manager._device_stats) == 0
+ assert len(result_manager._category_stats) == 0
+ assert len(result_manager._test_stats) == 0
+
+ # Add some test results
+ test1 = test_result_factory()
+ test1.name = "device1"
+ test1.result = AntaTestStatus.SUCCESS
+ test1.categories = ["system"]
+ test1.test = "test1"
+
+ test2 = test_result_factory()
+ test2.name = "device2"
+ test2.result = AntaTestStatus.FAILURE
+ test2.categories = ["interfaces"]
+ test2.test = "test2"
+
+ result_manager.add(test1)
+ result_manager.add(test2)
+
+ # Stats should still be unsynced after adding results
+ assert result_manager._stats_in_sync is False
+
+ # Test _compute_stats directly
+ with caplog.at_level(logging.INFO):
+ result_manager._compute_stats()
+ assert "Computing statistics for all results" in caplog.text
+ assert result_manager._stats_in_sync is True
+
+ # Verify stats content
+ assert len(result_manager._device_stats) == 2
+ assert len(result_manager._category_stats) == 2
+ assert len(result_manager._test_stats) == 2
+ assert result_manager._device_stats["device1"].tests_success_count == 1
+ assert result_manager._device_stats["device2"].tests_failure_count == 1
+ assert result_manager._category_stats["system"].tests_success_count == 1
+ assert result_manager._category_stats["interfaces"].tests_failure_count == 1
+ assert result_manager._test_stats["test1"].devices_success_count == 1
+ assert result_manager._test_stats["test2"].devices_failure_count == 1
+
+ def test_stats_property_computation(self, test_result_factory: Callable[[], TestResult], caplog: pytest.LogCaptureFixture) -> None:
+ """Test that stats are computed only once when accessed via properties."""
+ result_manager = ResultManager()
+
+ # Add some test results
+ test1 = test_result_factory()
+ test1.name = "device1"
+ test1.result = AntaTestStatus.SUCCESS
+ test1.categories = ["system"]
+ result_manager.add(test1)
+
+ test2 = test_result_factory()
+ test2.name = "device2"
+ test2.result = AntaTestStatus.FAILURE
+ test2.categories = ["interfaces"]
+ result_manager.add(test2)
+
+ # Stats should be unsynced after adding results
+ assert result_manager._stats_in_sync is False
+ assert "Computing statistics" not in caplog.text
+
+ # Access device_stats property - should trigger computation
+ with caplog.at_level(logging.INFO):
+ _ = result_manager.device_stats
+ assert "Computing statistics for all results" in caplog.text
+ assert result_manager._stats_in_sync is True
+
+ # Clear the log
+ caplog.clear()
+
+ # Access other stats properties - should not trigger computation again
+ with caplog.at_level(logging.INFO):
+ _ = result_manager.category_stats
+ _ = result_manager.test_stats
+ _ = result_manager.sorted_category_stats
+ assert "Computing statistics" not in caplog.text
+
+ # Add another result - should mark stats as unsynced
+ test3 = test_result_factory()
+ test3.name = "device3"
+ test3.result = "error"
+ result_manager.add(test3)
+ assert result_manager._stats_in_sync is False
+
+ # Access stats again - should trigger recomputation
+ with caplog.at_level(logging.INFO):
+ _ = result_manager.device_stats
+ assert "Computing statistics for all results" in caplog.text
+ assert result_manager._stats_in_sync is True