1"""
  2Pytest hooks for modifying tests as they run. Mainly used for modifying the HTML report.
  3
  4# (c) 2020-2024 TurnAround Factor, Inc.
  5#
  6# CUI DISTRIBUTION CONTROL
  7# Controlled by: DLA J68 R&D SBIP
  8# CUI Category: Small Business Research and Technology
  9# Distribution/Dissemination Controls: PROTECTED BY SBIR DATA RIGHTS
 10# POC: GOV SBIP Program Manager Denise Price, 571-767-0111
 11# Distribution authorized to U.S. Government Agencies only, to protect information not owned by the
 12# U.S. Government and protected by a contractor’s ‘limited rights’ statement, or received with the understanding that
 13# it is not routinely transmitted outside the U.S. Government (determination made September 14, 2024). Other requests
 14# for this document shall be referred to ACOR DLA Logistics Operations (J-68), 8725 John J. Kingman Rd., Suite 4317,
 15# Fort Belvoir, VA 22060-6221
 16#
 17# SBIR DATA RIGHTS
 18# Contract No.:SP4701-23-C-0083
 19# Contractor Name: TurnAround Factor, Inc.
 20# Contractor Address: 10365 Wood Park Ct. Suite 313 / Ashland, VA 23005
 21# Expiration of SBIR Data Rights Period: September 24, 2029
 22# The Government's rights to use, modify, reproduce, release, perform, display, or disclose technical data or computer
 23# software marked with this legend are restricted during the period shown as provided in paragraph (b)(4) of the Rights
 24# in Noncommercial Technical Data and Computer Software--Small Business Innovative Research (SBIR) Program clause
 25# contained in the above identified contract. No restrictions apply after the expiration date shown above. Any
 26# reproduction of technical data, computer software, or portions thereof marked with this legend must also reproduce
 27# the markings.
 28"""
 29
 30import html
 31import platform
 32import re
 33import signal
 34import sys
 35import time
 36from contextlib import suppress
 37from dataclasses import dataclass, field
 38from datetime import datetime, timedelta, timezone
 39from multiprocessing.managers import BaseManager
 40from pathlib import Path
 41
 42import pytest
 43from pytest_metadata.plugin import metadata_key
 44from hitl_tester.modules.cyber_6t.gpio import GPIOController
 45from hitl_tester.modules.cyber_6t.lcd_manager import PORT, AUTH_KEY
 46from hitl_tester.modules.cyber_6t.relay import RelayController
 47
 48from hitl_tester.modules import properties
 49from hitl_tester.modules.html_merger import Duration
 50from hitl_tester.modules.logger import logger
 51
 52PROFILE = ""
 53
 54properties.apply()
 55
 56
 57class SafeShutdown:
 58    """Manage shutting down safely."""
 59
 60    SIGNALS = [signal.SIGTSTP, signal.SIGQUIT, signal.SIGABRT, signal.SIGHUP, signal.SIGINT, signal.SIGTERM]
 61
 62    @staticmethod
 63    def handler(_signo, _stack_frame):
 64        """
 65        This handler will be called during normal / abnormal termination.
 66        i.e. ssh connection drops out. In this event, we're going to
 67        ensure that the test setup is transitioned to a safe state.
 68        """
 69        # pylint: disable=comparison-with-callable  # We want to check handler function
 70        # Failsafe: CTRL-Z sends kill
 71        signal.signal(signal.SIGTSTP, lambda _signo, _stack_frame: signal.raise_signal(signal.SIGKILL))
 72        if _signo:
 73            for signal_no in signal.valid_signals():  # Ignore all future signals if already shutting down
 74                if signal.getsignal(signal_no) == SafeShutdown.handler and signal_no != signal.SIGINT:
 75                    logger.write_info_to_report(f"Ignoring {signal.Signals(signal_no).name}")
 76                    signal.signal(signal_no, signal.SIG_IGN)
 77            logger.write_critical_to_report(f"Caught {signal.Signals(_signo).name}, Powering off HITL.")
 78        pytest.exit("HITL exiting test suite early", 1)  # Exit gracefully
 79
 80
 81class QueueManager(BaseManager):
 82    """Manager for multiprocess queue."""
 83
 84
 85@dataclass
 86class CaseStatus:
 87    """Holds information about the current test case."""
 88
 89    passed_tests: int = 0
 90    failed_tests: int = 0
 91    total_tests: int = 0
 92    completed_tests: int = 0
 93    test_ete: dict[str, int] = field(default_factory=dict)
 94    module_name: str = "Profile"
 95    mode: str = "Profile"
 96
 97
 98def get_test_name(item):
 99    """Make a unique name from an item."""
100    return f"{item.parent.name} - {item.originalname}"
101
102
103@pytest.hookimpl(trylast=True)
104def pytest_configure(config):
105    """Add information to the beginning of the test html report."""
106    config.addinivalue_line("markers", "profile: this test is used for generating a profile")
107
108    terminal_command = html.escape(" ".join(sys.argv))
109    hitl = html.escape(platform.node())
110    test_plan = html.escape(pytest.flags.test_plan.stem)
111    log_folder = html.escape(str(pytest.flags.report_filename.parent))
112    hardware_config = html.escape(pytest.flags.plateset_id)
113    properties_dict = {html.escape(name): html.escape(str(value)) for name, value in pytest.flags.properties.items()}
114
115    config.stash[metadata_key]["Terminal Command"] = f"<pre><code>{terminal_command}</code></pre>"
116    config.stash[metadata_key]["HITL"] = hitl
117    config.stash[metadata_key]["Test Plan"] = test_plan
118    config.stash[metadata_key]["Hardware Config"] = hardware_config
119    config.stash[metadata_key]["Properties"] = properties_dict
120    config.stash[metadata_key]["Logs"] = log_folder
121    config.stash[metadata_key]["Estimated Runtime (hh:mm:ss)"] = "Calculating..."
122
123    # Remove some rows
124    config.stash[metadata_key].pop("Plugins", None)
125    config.stash[metadata_key].pop("Packages", None)
126
127    # Rename some environment keys
128    config.stash[metadata_key]["Device Name"] = config.stash[metadata_key]["HITL"]
129    config.stash[metadata_key].pop("HITL", None)
130    config.stash[metadata_key]["Test Program"] = config.stash[metadata_key]["Test Plan"]
131    config.stash[metadata_key].pop("Test Plan", None)
132
133    # Move some environment keys to a details section
134    advanced_keys = ["Python", "Platform", "Terminal Command", "Hardware Config", "Properties"]
135    advanced_content = [f"{key}: {config.stash[metadata_key][key]}" for key in advanced_keys]
136    config.stash[metadata_key][
137        "Advanced Information"
138    ] = f"<details><summary>Expand</summary>{'<br>'.join(advanced_content)}</details>"
139
140    # Remove some rows
141    for key in advanced_keys:
142        config.stash[metadata_key].pop(key, None)
143
144
145def pytest_sessionstart(session):
146    """Setup variables."""
147    session.case_status = CaseStatus()
148    QueueManager.register("get_queue")
149    manager = QueueManager(address=("", PORT), authkey=AUTH_KEY)
150    session.queue = None
151    with suppress(OSError):
152        logger.write_info_to_report("Attempting to connect to LCD manager process...")
153        manager.connect()
154        session.queue = manager.get_queue()  # pylint: disable=no-member
155
156    for signal_flag in SafeShutdown.SIGNALS:
157        signal.signal(signal_flag, SafeShutdown.handler)
158
159    # Initialize relay and gpio for battery #1
160    RelayController.on(0)
161    GPIOController.battery_1.dormant_1 = False
162    GPIOController.battery_1.dormant_2 = False
163    GPIOController.battery_1.position_id = 0b1110
164    GPIOController.set_battery_state()
165    time.sleep(10)
166
167
168def pytest_collection_finish(session):
169    """Calculate the runtime for each test from their docstrings."""
170    session.case_status.total_tests = len(session.items)
171    total_estimated_runtime = timedelta()
172    for test in session.items:
173        time_match = (
174            re.search(
175                r"""
176            \|\s*Estimated\sDuration\s*\|\s*                 # Only look at the duration row
177            (\d+)                                            # Capture digits
178            \s+                                              # Space
179            (seconds* | minutes* | hours* | days* | weeks*)  # Capture unit of time (optional "s")
180            """,
181                test.function.__doc__,
182                re.IGNORECASE | re.VERBOSE,
183            )
184            if test.function.__doc__
185            else None
186        )
187        if time_match is not None:
188            unit_name = f"{time_match.group(2).rstrip('s')}s"
189            estimated_runtime = timedelta(**{unit_name: int(time_match.group(1))})
190            total_estimated_runtime += estimated_runtime
191            session.case_status.test_ete[get_test_name(test)] = estimated_runtime
192        else:
193            session.case_status.test_ete[get_test_name(test)] = timedelta()
194    session.config.stash[metadata_key]["Estimated Runtime (hh:mm:ss)"] = str(Duration(total_estimated_runtime))
195
196
197def pytest_runtest_setup(item):
198    """Send test status."""
199    name = item.parent.name.removeprefix("Test")
200    ete = sum(item.session.case_status.test_ete.values(), timedelta()).total_seconds() / 60
201    if item.parent.parent.name != "fingerprint.py":
202        item.session.case_status.mode = "PenTest"
203    else:
204        item.session.case_status.mode = "Profile" if not PROFILE else "Compare"
205    if item.session.queue:
206        item.session.queue.put(
207            {
208                "name": name,
209                "ete": ete,
210                "passed": item.session.case_status.passed_tests,
211                "failed": item.session.case_status.failed_tests,
212                "progress": item.session.case_status.completed_tests / item.session.case_status.total_tests,
213                "mode": item.session.case_status.mode,
214            }
215        )
216
217
218@pytest.hookimpl(tryfirst=True, hookwrapper=True)
219def pytest_runtest_makereport(item):
220    """Generate the report for one test."""
221    outcome = yield
222    report = outcome.get_result()
223    report.result_text = item.module.logger.html_result_text if hasattr(item.module, "logger") else ""
224    description_match = (
225        re.search(
226            r"""
227    \|\s*Description\s*\|\s*  # Description row
228    (.*?)                     # Capture description text
229    \s*\|                     # End of row
230    """,
231            item.function.__doc__,
232            re.IGNORECASE | re.VERBOSE,
233        )
234        if item.function.__doc__
235        else None
236    )
237    report.description = description_match.group(1) if description_match is not None else ""
238
239    # Modify pass/fail based on results
240    if (report.when == "setup" and report.skipped is True) or report.when == "call":
241        item.session.case_status.passed_tests += report.passed + report.skipped
242        item.session.case_status.failed_tests += report.failed
243        item.session.case_status.completed_tests += 1
244        item.session.case_status.test_ete.pop(get_test_name(item), None)
245
246
247def pytest_sessionfinish(session):
248    """Send final report status."""
249    if session.queue:
250        session.queue.put(
251            {
252                "name": "Test Completed",
253                "ete": 0.0,
254                "passed": session.case_status.passed_tests,
255                "failed": session.case_status.failed_tests,
256                "progress": 1.0,
257                "mode": session.case_status.mode,
258            }
259        )
260
261
262def pytest_html_results_table_row(report, cells):
263    """Generate the row for a test in the html report."""
264    del cells[:]
265
266    # Generate URL
267    test_case, line_number, test_name = report.location
268    test_name = test_name.partition("[")[0]  # Remove fixture information
269    test_case_html_path = str(Path(test_case).with_suffix(".html"))
270    base_link = html.escape(f"/docs/{test_case_html_path}#{test_name}")
271
272    # Generate result
273    pass_fail = report.outcome.title()
274    result = "<br>".join(report.result_text)
275    duration = timedelta(seconds=report.duration)
276    start_time = (datetime.now(timezone.utc) - duration).strftime("%Y/%m/%d %I:%M:%S %p")
277    safe_test_name = html.escape(test_name)
278    safe_test_case = html.escape(test_case)
279    cells.extend(
280        [
281            f'<td class="col-result">{pass_fail}</td>',
282            f"<td>{result}</td>",
283            f"<td>{report.description}</td>",
284            f'<td class="col-time">{start_time}</td>',
285            f'<td class="col-duration">{duration}</td>',
286            f'<td><a target="_blank" style="color:#337788;" href="{base_link}">{safe_test_name}</a> '
287            f"(File {safe_test_case}, line {line_number})</td>",
288        ]
289    )
290
291
292def pytest_html_results_table_header(cells):
293    """Generate html report headers."""
294    del cells[:]
295    cells.extend(
296        [
297            '<th class="sortable" data-column-type="result">Pass / Fail</th>',
298            "<th>Result</th>",
299            "<th>Description</th>",
300            '<th class="sortable time" data-column-type="time">Start Time</th>',
301            '<th class="sortable" data-column-type="duration">Duration (h:mm:ss.u)</th>',
302            '<th class="sortable" data-column-type="testId">Test / Docs</th>',
303        ]
304    )
305
306
307def pytest_html_report_title(report):
308    """Change the html report title."""
309    report.title = "HITL Report"