Source code for qolumbina.utils.save_tests

"""
The module provides utility functions to save test results of unit tests involved in 
``./tests/programs`` and parameters to JSON files.
"""

import json
from datetime import datetime
from typing import Dict, List, Literal, Tuple

import os
import pytest
import time
from typing import Any, Callable

def _save_test_result(
    test_obj: str,
    test_cases: dict[str, int],
    save_path: str,
):
    """
    Save test parameters and results to a JSON file
    in the same directory as the calling test file.

    Args:
        test_obj: The name of the test object (e.g., program name)
        test_cases: A dictionary containing test metadata and results
        save_path: The file path to save the JSON data
    """

    data = {
        "test_obj": test_obj,
        "time_stamp": datetime.now().isoformat(),
        "test_metadata": test_cases,
    }

    with open(save_path, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

    return save_path

[docs] def make_test_result_collector( program_name: str, save_dir: str, scope: Literal["session", "package", "module", "class", "function"] = "module" ): """ Factory to create a pytest fixture that collects test results and writes a summary JSON with pass_rate. Only stores id and result per test case. Args: program_name: Name of the program being tested, used in the JSON output save_dir: Directory where the JSON file will be saved scope: Scope of the pytest fixture (default: "module") """ def _fixture(): results: List[Dict] = [] def add_result(result: Dict): """ Expects result dict to contain at least: {"test_result": "PASS"/"FAIL"} Only id and test_result will be stored. """ test_result = result.get("test_result", "UNKNOWN") time_sec_val = float(result.get("time_sec", -1.0)) if time_sec_val < 0: # Time not recorded or invalid time_display = -1.0 elif time_sec_val < 0.1: # Less than 0.1 seconds, display "<0.1" time_display = "<0.1" else: # Round to 1 significant digit time_display = round(time_sec_val, 1) current_result = { "test_id": result.get("id", "N/A"), "test_result": test_result, "output_type": str(result.get("output_type", "N/A")), "execution_time_sec": time_display, "test_output": str(result.get("test_output", "N/A")), } if test_result == "FAIL": current_result["error"] = result.get("error", "No error message provided.") results.append(current_result) yield add_result, results # teardown: compute pass_rate & save JSON num_pass = sum(1 for r in results if r.get("test_result") == "PASS") total = len(results) pass_rate = round(num_pass / total, 3) if total > 0 else 0.0 # 3 decimal places summary = { "test_cases": results, "num_of_tests": total, "pass_rate": pass_rate } save_path = os.path.join(save_dir, f"results_{program_name}.json") # Save to JSON _save_test_result( test_obj=program_name, test_cases=summary, save_path=save_path ) # dynamically decorate fixture with scope return pytest.fixture(scope=scope)(_fixture)
[docs] def process_test_result( output_type: str, output_data: Any, request: pytest.FixtureRequest, test_oracle_func: Callable[[], None], test_start_time: float, test_result_collector: Tuple[Callable[[Dict[str, Any]], None], List[Dict[str, Any]]], ): """ Process and save the test result using the provided oracle function. Args: output_type: Type of the output (e.g., backend name). output_data: The actual output data from the test. request: Pytest fixture request object for accessing test metadata. test_oracle_func: A callable that performs assertions to validate the test. test_start_time: Timestamp when the test started. test_result_collector: A tuple containing a callable to add results and a list to store them. """ # Safely get test ID try: test_id = request.node.callspec.id except Exception: test_id = request.node.name test_result: dict[str, Any] = { "id": test_id, "output_type": output_type, } # Save test result try: test_oracle_func() test_result["test_result"] = "PASS" except AssertionError as e: test_result["test_result"] = "FAIL" test_result["error"] = str(e) raise finally: # Record time taken test_result["time_sec"] = time.time() - test_start_time try: if hasattr(output_data, "data"): test_result["test_output"] = output_data.data.tolist() else: test_result["test_output"] = output_data except Exception: test_result["test_output"] = "<unserializable>" # Safely add result to collector (must not affect pytest) try: test_result_collector[0](test_result) except Exception: pass