summaryrefslogtreecommitdiffstats
path: root/anta/reporter/__init__.py
blob: 3e068f5bd782193c54087fa0605875ec18b28e31 (plain)
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
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
# 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.
"""Report management for ANTA."""

# pylint: disable = too-few-public-methods
from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any

from jinja2 import Template
from rich.table import Table

from anta import RICH_COLOR_PALETTE, RICH_COLOR_THEME

if TYPE_CHECKING:
    import pathlib

    from anta.custom_types import TestStatus
    from anta.result_manager import ResultManager
    from anta.result_manager.models import TestResult

logger = logging.getLogger(__name__)


class ReportTable:
    """TableReport Generate a Table based on TestResult."""

    def _split_list_to_txt_list(self, usr_list: list[str], delimiter: str | None = None) -> str:
        """Split list to multi-lines string.

        Args:
        ----
            usr_list (list[str]): List of string to concatenate
            delimiter (str, optional): A delimiter to use to start string. Defaults to None.

        Returns
        -------
            str: Multi-lines string

        """
        if delimiter is not None:
            return "\n".join(f"{delimiter} {line}" for line in usr_list)
        return "\n".join(f"{line}" for line in usr_list)

    def _build_headers(self, headers: list[str], table: Table) -> Table:
        """Create headers for a table.

        First key is considered as header and is colored using RICH_COLOR_PALETTE.HEADER

        Args:
        ----
            headers: List of headers.
            table: A rich Table instance.

        Returns
        -------
            A rich `Table` instance with headers.

        """
        for idx, header in enumerate(headers):
            if idx == 0:
                table.add_column(header, justify="left", style=RICH_COLOR_PALETTE.HEADER, no_wrap=True)
            elif header == "Test Name":
                # We always want the full test name
                table.add_column(header, justify="left", no_wrap=True)
            else:
                table.add_column(header, justify="left")
        return table

    def _color_result(self, status: TestStatus) -> str:
        """Return a colored string based on the status value.

        Args:
        ----
            status (TestStatus): status value to color.

        Returns
        -------
        str: the colored string

        """
        color = RICH_COLOR_THEME.get(status, "")
        return f"[{color}]{status}" if color != "" else str(status)

    def report_all(self, manager: ResultManager, title: str = "All tests results") -> Table:
        """Create a table report with all tests for one or all devices.

        Create table with full output: Host / Test / Status / Message

        Args:
        ----
            manager: A ResultManager instance.
            title: Title for the report. Defaults to 'All tests results'.

        Returns
        -------
            A fully populated rich `Table`

        """
        table = Table(title=title, show_lines=True)
        headers = ["Device", "Test Name", "Test Status", "Message(s)", "Test description", "Test category"]
        table = self._build_headers(headers=headers, table=table)

        def add_line(result: TestResult) -> None:
            state = self._color_result(result.result)
            message = self._split_list_to_txt_list(result.messages) if len(result.messages) > 0 else ""
            categories = ", ".join(result.categories)
            table.add_row(str(result.name), result.test, state, message, result.description, categories)

        for result in manager.results:
            add_line(result)
        return table

    def report_summary_tests(
        self,
        manager: ResultManager,
        tests: list[str] | None = None,
        title: str = "Summary per test",
    ) -> Table:
        """Create a table report with result aggregated per test.

        Create table with full output: Test | Number of success | Number of failure | Number of error | List of nodes in error or failure

        Args:
        ----
            manager: A ResultManager instance.
            tests: List of test names to include. None to select all tests.
            title: Title of the report.

        Returns
        -------
            A fully populated rich `Table`.
        """
        table = Table(title=title, show_lines=True)
        headers = [
            "Test Case",
            "# of success",
            "# of skipped",
            "# of failure",
            "# of errors",
            "List of failed or error nodes",
        ]
        table = self._build_headers(headers=headers, table=table)
        for test in manager.get_tests():
            if tests is None or test in tests:
                results = manager.filter_by_tests({test}).results
                nb_failure = len([result for result in results if result.result == "failure"])
                nb_error = len([result for result in results if result.result == "error"])
                list_failure = [result.name for result in results if result.result in ["failure", "error"]]
                nb_success = len([result for result in results if result.result == "success"])
                nb_skipped = len([result for result in results if result.result == "skipped"])
                table.add_row(
                    test,
                    str(nb_success),
                    str(nb_skipped),
                    str(nb_failure),
                    str(nb_error),
                    str(list_failure),
                )
        return table

    def report_summary_devices(
        self,
        manager: ResultManager,
        devices: list[str] | None = None,
        title: str = "Summary per device",
    ) -> Table:
        """Create a table report with result aggregated per device.

        Create table with full output: Host | Number of success | Number of failure | Number of error | List of nodes in error or failure

        Args:
        ----
            manager: A ResultManager instance.
            devices: List of device names to include. None to select all devices.
            title: Title of the report.

        Returns
        -------
            A fully populated rich `Table`.
        """
        table = Table(title=title, show_lines=True)
        headers = [
            "Device",
            "# of success",
            "# of skipped",
            "# of failure",
            "# of errors",
            "List of failed or error test cases",
        ]
        table = self._build_headers(headers=headers, table=table)
        for device in manager.get_devices():
            if devices is None or device in devices:
                results = manager.filter_by_devices({device}).results
                nb_failure = len([result for result in results if result.result == "failure"])
                nb_error = len([result for result in results if result.result == "error"])
                list_failure = [result.test for result in results if result.result in ["failure", "error"]]
                nb_success = len([result for result in results if result.result == "success"])
                nb_skipped = len([result for result in results if result.result == "skipped"])
                table.add_row(
                    device,
                    str(nb_success),
                    str(nb_skipped),
                    str(nb_failure),
                    str(nb_error),
                    str(list_failure),
                )
        return table


class ReportJinja:
    """Report builder based on a Jinja2 template."""

    def __init__(self, template_path: pathlib.Path) -> None:
        """Create a ReportJinja instance."""
        if template_path.is_file():
            self.tempalte_path = template_path
        else:
            msg = f"template file is not found: {template_path}"
            raise FileNotFoundError(msg)

    def render(self, data: list[dict[str, Any]], *, trim_blocks: bool = True, lstrip_blocks: bool = True) -> str:
        """Build a report based on a Jinja2 template.

        Report is built based on a J2 template provided by user.
        Data structure sent to template is:

        >>> print(ResultManager.json)
        [
            {
                name: ...,
                test: ...,
                result: ...,
                messages: [...]
                categories: ...,
                description: ...,
            }
        ]

        Args:
        ----
            data: List of results from ResultManager.results
            trim_blocks: enable trim_blocks for J2 rendering.
            lstrip_blocks: enable lstrip_blocks for J2 rendering.

        Returns
        -------
            Rendered template

        """
        with self.tempalte_path.open(encoding="utf-8") as file_:
            template = Template(file_.read(), trim_blocks=trim_blocks, lstrip_blocks=lstrip_blocks)

        return template.render({"data": data})