1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
|
# Copyright (c) 2023-2024 Arista Networks, Inc.
# Use of this source code is governed by the Apache License 2.0
# that can be found in the LICENSE file.
"""
Test anta.result_manager.__init__.py
"""
from __future__ import annotations
import json
from contextlib import nullcontext
from typing import TYPE_CHECKING, Any, Callable
import pytest
from anta.custom_types import TestStatus
from anta.result_manager import ResultManager
if TYPE_CHECKING:
from anta.result_manager.models import TestResult
class Test_ResultManager:
"""
Test ResultManager class
"""
# not testing __init__ as nothing is going on there
def test__len__(self, list_result_factory: Callable[[int], list[TestResult]]) -> None:
"""
test __len__
"""
list_result = list_result_factory(3)
result_manager = ResultManager()
assert len(result_manager) == 0
for i in range(3):
result_manager.add_test_result(list_result[i])
assert len(result_manager) == i + 1
@pytest.mark.parametrize(
"starting_status, test_status, expected_status, expected_raise",
[
pytest.param("unset", "unset", "unset", nullcontext(), id="unset->unset"),
pytest.param("unset", "success", "success", nullcontext(), id="unset->success"),
pytest.param("unset", "error", "unset", nullcontext(), id="set error"),
pytest.param("skipped", "skipped", "skipped", nullcontext(), id="skipped->skipped"),
pytest.param("skipped", "unset", "skipped", nullcontext(), id="skipped, add unset"),
pytest.param("skipped", "success", "success", nullcontext(), id="skipped, add success"),
pytest.param("skipped", "failure", "failure", nullcontext(), id="skipped, add failure"),
pytest.param("success", "unset", "success", nullcontext(), id="success, add unset"),
pytest.param("success", "skipped", "success", nullcontext(), id="success, add skipped"),
pytest.param("success", "success", "success", nullcontext(), id="success->success"),
pytest.param("success", "failure", "failure", nullcontext(), id="success->failure"),
pytest.param("failure", "unset", "failure", nullcontext(), id="failure->failure"),
pytest.param("failure", "skipped", "failure", nullcontext(), id="failure, add unset"),
pytest.param("failure", "success", "failure", nullcontext(), id="failure, add skipped"),
pytest.param("failure", "failure", "failure", nullcontext(), id="failure, add success"),
pytest.param("unset", "unknown", None, pytest.raises(ValueError), id="wrong status"),
],
)
def test__update_status(self, starting_status: TestStatus, test_status: TestStatus, expected_status: str, expected_raise: Any) -> None:
"""
Test ResultManager._update_status
"""
result_manager = ResultManager()
result_manager.status = starting_status
assert result_manager.error_status is False
with expected_raise:
result_manager._update_status(test_status) # pylint: disable=protected-access
if test_status == "error":
assert result_manager.error_status is True
else:
assert result_manager.status == expected_status
def test_add_test_result(self, test_result_factory: Callable[[int], TestResult]) -> None:
"""
Test ResultManager.add_test_result
"""
result_manager = ResultManager()
assert result_manager.status == "unset"
assert result_manager.error_status is False
assert len(result_manager) == 0
# Add one unset test
unset_test = test_result_factory(0)
unset_test.result = "unset"
result_manager.add_test_result(unset_test)
assert result_manager.status == "unset"
assert result_manager.error_status is False
assert len(result_manager) == 1
# Add one success test
success_test = test_result_factory(1)
success_test.result = "success"
result_manager.add_test_result(success_test)
assert result_manager.status == "success"
assert result_manager.error_status is False
assert len(result_manager) == 2
# Add one error test
error_test = test_result_factory(1)
error_test.result = "error"
result_manager.add_test_result(error_test)
assert result_manager.status == "success"
assert result_manager.error_status is True
assert len(result_manager) == 3
# Add one failure test
failure_test = test_result_factory(1)
failure_test.result = "failure"
result_manager.add_test_result(failure_test)
assert result_manager.status == "failure"
assert result_manager.error_status is True
assert len(result_manager) == 4
def test_add_test_results(self, list_result_factory: Callable[[int], list[TestResult]]) -> None:
"""
Test ResultManager.add_test_results
"""
result_manager = ResultManager()
assert result_manager.status == "unset"
assert result_manager.error_status is False
assert len(result_manager) == 0
# Add three success tests
success_list = list_result_factory(3)
for test in success_list:
test.result = "success"
result_manager.add_test_results(success_list)
assert result_manager.status == "success"
assert result_manager.error_status is False
assert len(result_manager) == 3
# Add one error test and one failure
error_failure_list = list_result_factory(2)
error_failure_list[0].result = "error"
error_failure_list[1].result = "failure"
result_manager.add_test_results(error_failure_list)
assert result_manager.status == "failure"
assert result_manager.error_status is True
assert len(result_manager) == 5
@pytest.mark.parametrize(
"status, error_status, ignore_error, expected_status",
[
pytest.param("success", False, True, "success", id="no error"),
pytest.param("success", True, True, "success", id="error, ignore error"),
pytest.param("success", True, False, "error", id="error, do not ignore error"),
],
)
def test_get_status(self, status: TestStatus, error_status: bool, ignore_error: bool, expected_status: str) -> None:
"""
test ResultManager.get_status
"""
result_manager = ResultManager()
result_manager.status = status
result_manager.error_status = error_status
assert result_manager.get_status(ignore_error=ignore_error) == expected_status
def test_get_results(self, list_result_factory: Callable[[int], list[TestResult]]) -> None:
"""
test ResultManager.get_results
"""
result_manager = ResultManager()
success_list = list_result_factory(3)
for test in success_list:
test.result = "success"
result_manager.add_test_results(success_list)
res = result_manager.get_results()
assert isinstance(res, list)
def test_get_json_results(self, list_result_factory: Callable[[int], list[TestResult]]) -> None:
"""
test ResultManager.get_json_results
"""
result_manager = ResultManager()
success_list = list_result_factory(3)
for test in success_list:
test.result = "success"
result_manager.add_test_results(success_list)
json_res = result_manager.get_json_results()
assert isinstance(json_res, str)
# Verifies it can be deserialized back to a list of dict with the correct values types
res = json.loads(json_res)
for test in res:
assert isinstance(test, dict)
assert isinstance(test.get("test"), str)
assert isinstance(test.get("categories"), list)
assert isinstance(test.get("description"), str)
assert test.get("custom_field") is None
assert test.get("result") == "success"
# TODO
# get_result_by_test
# get_result_by_host
# get_testcases
# get_hosts
|
