diff --git a/src/imgtests/constant.py b/src/imgtests/constant.py index 2d87f38d..371ea115 100644 --- a/src/imgtests/constant.py +++ b/src/imgtests/constant.py @@ -14,6 +14,11 @@ REPORTS_DIR: Final = LIB_DATA_DIR / "results" EXCEL_REPORTS_DIR: Final = LIB_DATA_DIR / "excel_reports" +DISTRIBUTION_DESCRIPTIONS: Final[dict[str, str]] = { + "poky": "Poky Linux distribution", + "suse": "SUSE Linux distribution", +} + SSH_CLIENT_MISSING_RESULT: Final = TestResult( status=TestStatus.BROKEN, # TODO: use frozendict from 3.15 diff --git a/src/imgtests/reporting/excel_export.py b/src/imgtests/reporting/excel_export.py index fa4458f6..cf535a23 100644 --- a/src/imgtests/reporting/excel_export.py +++ b/src/imgtests/reporting/excel_export.py @@ -7,12 +7,13 @@ from collections.abc import Callable, Iterable, Mapping, Sequence from datetime import date, datetime from pathlib import Path -from typing import TYPE_CHECKING, Any, Final +from typing import TYPE_CHECKING, Any, Final, TypedDict from openpyxl import Workbook from sqlalchemy import create_engine, inspect, select from sqlalchemy.orm import Session +from imgtests.constant import DISTRIBUTION_DESCRIPTIONS from imgtests.database.models.configuration import ConfigurationBase from imgtests.database.models.experiment import ExperimentBase from imgtests.database.models.util_run_result import UtilRunResult @@ -29,18 +30,9 @@ TABLES: Final = ("experiment", "util_run_result") logger = logging.getLogger(__name__) CONFIGURATION_SHEET: Final = "configuration" +DEFAULT_OUTPUT_FILENAME: Final = "report.xlsx" DATABASE_COMMAND: Final = "database" DISTRO_COMPARISON_COMMAND: Final = "distro-comparison" -DISTRIBUTIONS: Final[dict[str, dict[str, int | str]]] = { - "poky": { - "configuration_id": 1, - "distribution_description": "Poky Linux distribution", - }, - "suse": { - "configuration_id": 2, - "distribution_description": "SUSE Linux distribution", - }, -} JSON_COLUMNS: Final = { "util_run_result": {"result"}, @@ -130,13 +122,29 @@ def add_database_arguments(parser: argparse.ArgumentParser) -> None: default=list(TABLES), help="Tables to export. Defaults to all project result tables.", ) + parser.add_argument( + "--configuration-id", + dest="configuration_ids", + action="append", + required=True, + metavar="DISTRO=ID", + type=parse_configuration_id_override, + help=( + "Configuration id to write for a distribution. " + "Must be passed for each supported distribution, for example: " + "--configuration-id poky=10 --configuration-id suse=20." + ), + ) def export_database_to_excel( engine: Engine, output_path: Path, + configuration_ids: dict[str, int], tables: Sequence[str] = tuple(TABLES), -) -> None: +) -> Path: + output_path = prepare_output_path(output_path) + distributions = build_distribution_records(configuration_ids) unsupported_tables = [table for table in tables if table not in TABLES] if unsupported_tables: @@ -166,12 +174,12 @@ def export_database_to_excel( worksheets.append( flatten_table_records( CONFIGURATION_SHEET, - fetch_configuration_records(engine), + fetch_configuration_records(engine, distributions), ), ) for table in tables: - records = fetch_table_records(engine, table) + records = fetch_table_records(engine, table, distributions) if table == "util_run_result": worksheets.extend(flatten_util_run_results(records)) @@ -179,20 +187,28 @@ def export_database_to_excel( worksheets.append(flatten_table_records(table, records)) write_xlsx(output_path, worksheets) + return output_path -def fetch_table_records(engine: Engine, table: str) -> list[dict[str, Any]]: +def fetch_table_records( + engine: Engine, + table: str, + distributions: Mapping[str, DistributionRecord], +) -> list[dict[str, Any]]: if table == "experiment": - return fetch_experiment_records(engine) + return fetch_experiment_records(engine, distributions) if table == "util_run_result": - return fetch_util_run_result_records(engine) + return fetch_util_run_result_records(engine, distributions) msg = f"Unsupported export table: {table}" raise ValueError(msg) -def fetch_experiment_records(engine: Engine) -> list[dict[str, Any]]: +def fetch_experiment_records( + engine: Engine, + distributions: Mapping[str, DistributionRecord], +) -> list[dict[str, Any]]: query = ( select( ConfigurationBase.os.label("configuration_os"), @@ -215,10 +231,13 @@ def fetch_experiment_records(engine: Engine) -> list[dict[str, Any]]: with Session(engine) as session: result = session.execute(query) - return [add_configuration_id(dict(row)) for row in result.mappings()] + return [add_configuration_id(dict(row), distributions) for row in result.mappings()] -def fetch_util_run_result_records(engine: Engine) -> list[dict[str, Any]]: +def fetch_util_run_result_records( + engine: Engine, + distributions: Mapping[str, DistributionRecord], +) -> list[dict[str, Any]]: query = ( select( ConfigurationBase.os.label("configuration_os"), @@ -239,25 +258,106 @@ def fetch_util_run_result_records(engine: Engine) -> list[dict[str, Any]]: with Session(engine) as session: result = session.execute(query) - return [add_configuration_id(dict(row)) for row in result.mappings()] + return [add_configuration_id(dict(row), distributions) for row in result.mappings()] -def fetch_configuration_records(_engine: Engine) -> list[dict[str, Any]]: - return list(DISTRIBUTIONS.values()) +def fetch_configuration_records( + _engine: Engine, + distributions: Mapping[str, DistributionRecord], +) -> list[dict[str, Any]]: + return [dict(record) for record in distributions.values()] -def add_configuration_id(record: dict[str, Any]) -> dict[str, Any]: +def add_configuration_id( + record: dict[str, Any], + distributions: Mapping[str, DistributionRecord], +) -> dict[str, Any]: os_name = record.pop("configuration_os", None) - return {"configuration_id": configuration_id_from_os(os_name), **record} + return {"configuration_id": configuration_id_from_os(os_name, distributions), **record} -def configuration_id_from_os(os_name: Any) -> int | str: +def configuration_id_from_os( + os_name: Any, + distributions: Mapping[str, DistributionRecord], +) -> int | str: distribution_name = distribution_name_from_os(os_name) if not distribution_name: return "" - return int(DISTRIBUTIONS[distribution_name]["configuration_id"]) + return int(distributions[distribution_name]["configuration_id"]) + + +class DistributionRecord(TypedDict): + configuration_id: int + distribution_description: str + + +def build_distribution_records( + configuration_ids: dict[str, int], +) -> dict[str, DistributionRecord]: + ids = { + distribution_name.strip().lower(): configuration_id + for distribution_name, configuration_id in configuration_ids.items() + } + unsupported_distributions = [ + distribution_name + for distribution_name in ids + if distribution_name not in DISTRIBUTION_DESCRIPTIONS + ] + + if unsupported_distributions: + valid_names = ", ".join(DISTRIBUTION_DESCRIPTIONS) + joined_distributions = ", ".join(unsupported_distributions) + msg = f"Unsupported distributions: {joined_distributions}. Available: {valid_names}." + raise ValueError(msg) + + missing_distributions = [ + distribution_name + for distribution_name in DISTRIBUTION_DESCRIPTIONS + if distribution_name not in ids + ] + + if missing_distributions: + joined_distributions = ", ".join(missing_distributions) + msg = f"Missing configuration ids for distributions: {joined_distributions}." + raise ValueError(msg) + + return { + distribution_name: DistributionRecord( + configuration_id=ids[distribution_name], + distribution_description=distribution_description, + ) + for distribution_name, distribution_description in DISTRIBUTION_DESCRIPTIONS.items() + } + + +def parse_configuration_id_override(value: str) -> tuple[str, int]: + distribution_name, separator, raw_id = value.partition("=") + + if not separator: + msg = "Expected configuration id in DISTRO=ID format." + raise argparse.ArgumentTypeError(msg) + + distribution_name = distribution_name.strip().lower() + raw_id = raw_id.strip() + + if distribution_name not in DISTRIBUTION_DESCRIPTIONS: + valid_names = ", ".join(DISTRIBUTION_DESCRIPTIONS) + msg = f"Unsupported distribution '{distribution_name}'. Available: {valid_names}." + raise argparse.ArgumentTypeError(msg) + + try: + configuration_id = int(raw_id) + except ValueError as exc: + msg = f"Configuration id for '{distribution_name}' must be an integer." + raise argparse.ArgumentTypeError(msg) from exc + + if configuration_id <= 0: + msg = f"Configuration id for '{distribution_name}' must be positive." + raise argparse.ArgumentTypeError(msg) + + return distribution_name, configuration_id def distribution_name_from_os(os_name: Any) -> str: @@ -359,12 +459,11 @@ def flatten_util_run_result(record: Mapping[str, Any]) -> dict[str, Any]: tool_name = detect_utility(result) record_without_configuration = dict(record) - row = { + row: dict[str, str] = { "configuration_id": record_without_configuration.pop("configuration_id", ""), "test_name": test_name, "tool_name": tool_name, } - row.update( flatten_record_with_column_names( record_without_configuration, @@ -483,11 +582,10 @@ def util_run_result_test_name(record: Mapping[str, Any], result: Any) -> str: def result_based_test_name(result: Any) -> str: - if not isinstance(result, Mapping): + if not isinstance(result, dict): return "" - tool = str(result.get("tool") or "").strip() - + tool = str(result.get("tool", "")).strip() if tool: return result_tool_test_name(tool, result.get("test_type")) @@ -495,12 +593,11 @@ def result_based_test_name(result: Any) -> str: def result_tool_test_name(tool: str, test_type: Any) -> str: - if not isinstance(test_type, Mapping): + if not isinstance(test_type, dict): return tool for field in ("name", "stressor", "protocol"): - value = str(test_type.get(field) or "").strip() - + value = str(test_type.get(field, "")).strip() if value and value != "unknown": return f"{tool}_{value}" @@ -508,14 +605,14 @@ def result_tool_test_name(tool: str, test_type: Any) -> str: def detect_utility(result: Any) -> str: - if isinstance(result, Mapping): - return str(result.get("tool") or "").strip() + if isinstance(result, dict): + return str(result.get("tool", "")).strip() return "" def command_name_from_record(record: Mapping[str, Any]) -> str: - command = str(record.get("command") or "").strip() + command = str(record.get("command", "")).strip() if not command: return "" @@ -548,34 +645,28 @@ def readable_util_column_name(key: str) -> str: def normalize_metric_column(path: str) -> str: - parts = [] - + parts: list[str] = [] for raw_part in column_path_parts(path): if raw_part.isdecimal(): continue - part = COLUMN_PART_REPLACEMENTS.get(raw_part, raw_part) - if part in NOISY_COLUMN_PARTS: continue - if parts and parts[-1] == part: continue - parts.append(part) return "_".join(parts) or "value" def column_path_parts(path: str) -> list[str]: - parts = [] - current = [] + parts: list[str] = [] + current: list[str] = [] for char in path: if char.isalnum() or char == "_": current.append(char.lower()) continue - if current: parts.append("".join(current)) current.clear() @@ -606,12 +697,11 @@ def flatten_json(prefix: str, value: Any) -> list[tuple[str, Any]]: if decoded_value is not value: return flatten_json(prefix, decoded_value) - if isinstance(value, Mapping): + if isinstance(value, dict): if not value: return [(prefix, None)] flattened: list[tuple[str, Any]] = [] - for key, nested_value in value.items(): nested_prefix = f"{prefix}.{key}" flattened.extend(flatten_json(nested_prefix, nested_value)) @@ -671,7 +761,7 @@ def __init__(self, name: str, headers: list[str], rows: list[dict[str, Any]]) -> def write_xlsx(output_path: Path, worksheets: Sequence[WorksheetData]) -> None: - output_path.parent.mkdir(parents=True, exist_ok=True) + output_path = prepare_output_path(output_path) sheet_names = unique_sheet_names(worksheet.name for worksheet in worksheets) workbook = Workbook() @@ -687,6 +777,38 @@ def write_xlsx(output_path: Path, worksheets: Sequence[WorksheetData]) -> None: workbook.save(output_path) +def prepare_output_path(output_path: Path) -> Path: + output_path = output_path.expanduser() + + if (output_path.exists() and output_path.is_dir()) or not output_path.suffix: + output_path /= DEFAULT_OUTPUT_FILENAME + elif output_path.suffix.lower() != ".xlsx": + msg = f"Output file must have .xlsx extension: {output_path}" + raise ValueError(msg) + + output_parent = output_path.parent + + if output_parent.exists() and not output_parent.is_dir(): + msg = f"Output parent path is not a directory: {output_parent}" + raise ValueError(msg) + + try: + output_parent.mkdir(parents=True, exist_ok=True) + except OSError as exc: + msg = f"Cannot create output directory '{output_parent}': {exc}" + raise ValueError(msg) from exc + + if output_path.exists() and output_path.is_dir(): + msg = f"Output path is a directory: {output_path}" + raise ValueError(msg) + + if output_path.exists() and not output_path.is_file(): + msg = f"Output path is not a regular file: {output_path}" + raise ValueError(msg) + + return output_path + + def unique_sheet_names(names: Iterable[str]) -> list[str]: used_names: set[str] = set() result: list[str] = [] @@ -736,13 +858,17 @@ def main() -> None: engine = create_engine(args.db_url) - export_database_to_excel( - engine=engine, - output_path=args.output, - tables=args.tables, - ) + try: + output_path = export_database_to_excel( + engine=engine, + output_path=args.output, + tables=args.tables, + configuration_ids=dict(args.configuration_ids), + ) + except ValueError as exc: + raise SystemExit(str(exc)) from exc - logger.info("Exported tables %s to %s", ", ".join(args.tables), args.output) + logger.info("Exported tables %s to %s", ", ".join(args.tables), output_path) def export_distro_comparison_command(args: argparse.Namespace) -> None: diff --git a/src/imgtests/web/tests_interface/views.py b/src/imgtests/web/tests_interface/views.py index 059ed572..04a24e34 100644 --- a/src/imgtests/web/tests_interface/views.py +++ b/src/imgtests/web/tests_interface/views.py @@ -383,7 +383,11 @@ def api_export_excel(request: HttpRequest) -> JsonResponse: # noqa: ARG001 try: engine = create_engine(db_url) - export_database_to_excel(engine=engine, output_path=output_path) + export_database_to_excel( + engine=engine, + output_path=output_path, + configuration_ids={"poky": 1, "suse": 2}, + ) except Exception as err: # noqa: BLE001 return JsonResponse({"error": f"Export failed: {err}"}, status=500)