diff --git a/tests/test_run_tests.py b/tests/test_run_tests.py index 3761518..1dc400e 100644 --- a/tests/test_run_tests.py +++ b/tests/test_run_tests.py @@ -593,7 +593,7 @@ def test_run_tests_logger_configuration( # Assert assert result.exit_code == 0 - mock_configure_logger.assert_called_once_with(title="Custom Logger Test") + mock_configure_logger.assert_called_once_with(title="Custom Logger Test", enable_log_streaming=True) assert "Log output in: /path/to/test_logs/custom_run.log" in result.output def test_run_tests_default_title_generation( diff --git a/tests/test_tc_params_mapping.py b/tests/test_tc_params_mapping.py new file mode 100644 index 0000000..b3d7926 --- /dev/null +++ b/tests/test_tc_params_mapping.py @@ -0,0 +1,373 @@ +# +# Copyright (c) 2026 Project CHIP Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +"""Unit tests for load_tc_params_mapping (utils), validate_tc_params_file, and +parse_and_validate_tc_params_file (validation).""" + +import json +from pathlib import Path + +import pytest + +from th_cli.exceptions import CLIError +from th_cli.utils import load_tc_params_mapping +from th_cli.validation import parse_and_validate_tc_params_file, validate_tc_params_file + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _write_mapping(tmp_path: Path, data: dict, filename: str = "mapping.json") -> Path: + """Write *data* as JSON to *tmp_path / filename* and return the path.""" + p = tmp_path / filename + p.write_text(json.dumps(data), encoding="utf-8") + return p + + +# --------------------------------------------------------------------------- +# parse_and_validate_tc_params_file (shared helper) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestParseAndValidateTcParamsFile: + """Tests for parse_and_validate_tc_params_file — the shared parsing helper. + + Because both validate_tc_params_file and load_tc_params_mapping delegate + to this function, its error paths are exercised transitively by those + tests too. These tests focus on the return value and the fact that the + helper is the single source of truth. + """ + + def test_returns_parsed_dict(self, tmp_path: Path) -> None: + """Helper returns the full parsed mapping as a dict.""" + data = {"TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}} + p = _write_mapping(tmp_path, data) + result = parse_and_validate_tc_params_file(str(p)) + assert result == data + + def test_empty_mapping_returns_empty_dict(self, tmp_path: Path) -> None: + """An empty JSON object {} returns an empty dict without error.""" + p = _write_mapping(tmp_path, {}) + assert parse_and_validate_tc_params_file(str(p)) == {} + + def test_file_not_found_raises(self, tmp_path: Path) -> None: + """Missing file raises CLIError.""" + with pytest.raises(CLIError, match="File not found"): + parse_and_validate_tc_params_file(str(tmp_path / "missing.json")) + + def test_invalid_json_raises(self, tmp_path: Path) -> None: + """Broken JSON raises CLIError with line/column info.""" + p = tmp_path / "bad.json" + p.write_text("{bad", encoding="utf-8") + with pytest.raises(CLIError, match="Invalid JSON"): + parse_and_validate_tc_params_file(str(p)) + + def test_array_root_raises(self, tmp_path: Path) -> None: + """A JSON array at root raises CLIError.""" + p = tmp_path / "arr.json" + p.write_text("[]", encoding="utf-8") + with pytest.raises(CLIError, match="Expected a JSON object at the top level"): + parse_and_validate_tc_params_file(str(p)) + + def test_non_dict_entry_value_raises(self, tmp_path: Path) -> None: + """A non-dict entry value raises CLIError naming the offending TC ID.""" + p = _write_mapping(tmp_path, {"TC-ACE-1.1": "string-not-dict"}) + with pytest.raises(CLIError, match="TC-ACE-1.1"): + parse_and_validate_tc_params_file(str(p)) + + def test_validate_and_load_share_same_errors(self, tmp_path: Path) -> None: + """validate_tc_params_file and load_tc_params_mapping raise the same + CLIError message for the same bad file — proving they share the helper.""" + p = _write_mapping(tmp_path, {"TC-ACE-1.1": 42}) + + with pytest.raises(CLIError) as exc_validate: + validate_tc_params_file(str(p)) + + with pytest.raises(CLIError) as exc_load: + load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + + assert str(exc_validate.value) == str(exc_load.value) + + +# --------------------------------------------------------------------------- +# validate_tc_params_file +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestValidateTcParamsFile: + """Tests for validate_tc_params_file.""" + + def test_valid_mapping_file_returns_path(self, tmp_path: Path) -> None: + """A well-formed mapping file returns its resolved Path.""" + p = _write_mapping(tmp_path, {"TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}}) + result = validate_tc_params_file(str(p)) + assert result.is_file() + + def test_empty_mapping_is_valid(self, tmp_path: Path) -> None: + """An empty JSON object {} is a valid (but empty) mapping file.""" + p = _write_mapping(tmp_path, {}) + result = validate_tc_params_file(str(p)) + assert result.is_file() + + def test_file_not_found_raises(self, tmp_path: Path) -> None: + """A non-existent path raises CLIError.""" + missing = tmp_path / "no_such_mapping.json" + with pytest.raises(CLIError, match="File not found"): + validate_tc_params_file(str(missing)) + + def test_directory_instead_of_file_raises(self, tmp_path: Path) -> None: + """Passing a directory raises CLIError.""" + with pytest.raises(CLIError, match="not a file"): + validate_tc_params_file(str(tmp_path)) + + def test_invalid_json_raises(self, tmp_path: Path) -> None: + """A file with broken JSON raises CLIError mentioning 'Invalid JSON'.""" + p = tmp_path / "bad.json" + p.write_text('{"TC-ACE-1.1": {', encoding="utf-8") + with pytest.raises(CLIError, match="Invalid JSON"): + validate_tc_params_file(str(p)) + + def test_json_array_root_raises(self, tmp_path: Path) -> None: + """A JSON array at the top level raises CLIError.""" + p = tmp_path / "array.json" + p.write_text('[{"int-arg": "x"}]', encoding="utf-8") + with pytest.raises(CLIError, match="Expected a JSON object at the top level"): + validate_tc_params_file(str(p)) + + def test_non_dict_value_raises(self, tmp_path: Path) -> None: + """An entry whose value is not a dict raises CLIError listing the bad TC ID.""" + p = _write_mapping(tmp_path, {"TC-ACE-1.1": "not-a-dict"}) + with pytest.raises(CLIError, match="TC-ACE-1.1"): + validate_tc_params_file(str(p)) + + def test_multiple_non_dict_values_listed_in_error(self, tmp_path: Path) -> None: + """All bad TC IDs are listed in the error message.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": 42, + "TC-CC-1.1": ["list", "value"], + "TC-OK-1.1": {"int-arg": "PIXIT.OK:1"}, + }) + with pytest.raises(CLIError) as exc_info: + validate_tc_params_file(str(p)) + msg = str(exc_info.value) + assert "TC-ACE-1.1" in msg + assert "TC-CC-1.1" in msg + assert "TC-OK-1.1" not in msg # valid entry must NOT appear in error + + def test_valid_multi_entry_mapping(self, tmp_path: Path) -> None: + """Multiple valid entries with varied parameter keys are all accepted.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1", "timeout": "300"}, + "TC-AVSM-2.10": {"int-arg": "PIXIT.AVSM.EP:1 PIXIT.AVSM.ST:1"}, + "TC-MCORE-FS-1.3": {"string-arg": "PIXIT.MCORE.EP:3"}, + }) + result = validate_tc_params_file(str(p)) + assert result.exists() + + +# --------------------------------------------------------------------------- +# load_tc_params_mapping +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestLoadTcParamsMapping: + """Tests for load_tc_params_mapping.""" + + # ------------------------------------------------------------------ + # Happy-path: matching behaviour + # ------------------------------------------------------------------ + + def test_exact_match_returns_params(self, tmp_path: Path) -> None: + """A TC ID that matches exactly returns its params.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + assert params == {"int-arg": "PIXIT.ACE.EP:1"} + assert missing == [] + + def test_dash_underscore_normalisation(self, tmp_path: Path) -> None: + """Mapping key uses dashes; caller uses underscores — both normalise the same.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC_ACE_1_1"]) + assert params == {"int-arg": "PIXIT.ACE.EP:1"} + assert missing == [] + + def test_dot_normalisation(self, tmp_path: Path) -> None: + """Dots in TC IDs are treated the same as dashes and underscores.""" + p = _write_mapping(tmp_path, { + "TC.ACE.1.1": {"timeout": "300"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + assert params == {"timeout": "300"} + assert missing == [] + + def test_case_insensitive_match(self, tmp_path: Path) -> None: + """Keys and test IDs are compared case-insensitively.""" + p = _write_mapping(tmp_path, { + "tc-ace-1.1": {"int-arg": "PIXIT.ACE.EP:1"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + assert params == {"int-arg": "PIXIT.ACE.EP:1"} + assert missing == [] + + def test_multiple_ids_merged(self, tmp_path: Path) -> None: + """Params from multiple matched TC IDs are all merged into one dict.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}, + "TC-CC-1.1": {"timeout": "600"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1", "TC-CC-1.1"]) + assert params == {"int-arg": "PIXIT.ACE.EP:1", "timeout": "600"} + assert missing == [] + + def test_unmatched_ids_reported_in_missing(self, tmp_path: Path) -> None: + """TC IDs absent from the mapping are reported in the missing list.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1", "TC-NONEXISTENT-1.1"]) + assert params == {"int-arg": "PIXIT.ACE.EP:1"} + assert "TC-NONEXISTENT-1.1" in missing + + def test_all_ids_unmatched(self, tmp_path: Path) -> None: + """When no IDs match, merged_params is empty and all IDs are in missing.""" + p = _write_mapping(tmp_path, {"TC-CC-1.1": {"timeout": "300"}}) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + assert params == {} + assert missing == ["TC-ACE-1.1"] + + def test_empty_mapping_all_missing(self, tmp_path: Path) -> None: + """An empty mapping file yields an empty params dict and all IDs as missing.""" + p = _write_mapping(tmp_path, {}) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1", "TC-CC-1.1"]) + assert params == {} + assert set(missing) == {"TC-ACE-1.1", "TC-CC-1.1"} + + def test_empty_test_ids_list(self, tmp_path: Path) -> None: + """An empty test_ids list yields empty results without error.""" + p = _write_mapping(tmp_path, {"TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}}) + params, missing = load_tc_params_mapping(str(p), []) + assert params == {} + assert missing == [] + + # ------------------------------------------------------------------ + # Merge-order: last match wins on key conflicts + # ------------------------------------------------------------------ + + def test_key_conflict_last_sorted_id_wins(self, tmp_path: Path) -> None: + """When two matched entries share a key, the alphabetically later TC ID wins.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1", "timeout": "100"}, + "TC-CC-1.1": {"int-arg": "PIXIT.CC.EP:2", "timeout": "999"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1", "TC-CC-1.1"]) + # Sorted alphabetically: TC-ACE-1.1 first, TC-CC-1.1 second → CC wins + assert params["int-arg"] == "PIXIT.CC.EP:2" + assert params["timeout"] == "999" + assert missing == [] + + def test_non_conflicting_keys_all_present(self, tmp_path: Path) -> None: + """Disjoint parameter keys from multiple TCs are all present in result.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}, + "TC-CC-1.1": {"string-arg": "PIXIT.CC.NAME:foo"}, + "TC-ICT-1.1": {"timeout": "500"}, + }) + params, _ = load_tc_params_mapping(str(p), ["TC-ACE-1.1", "TC-CC-1.1", "TC-ICT-1.1"]) + assert params["int-arg"] == "PIXIT.ACE.EP:1" + assert params["string-arg"] == "PIXIT.CC.NAME:foo" + assert params["timeout"] == "500" + + # ------------------------------------------------------------------ + # Error paths + # ------------------------------------------------------------------ + + def test_file_not_found_raises(self, tmp_path: Path) -> None: + """A non-existent mapping file raises CLIError.""" + missing_file = str(tmp_path / "no_file.json") + with pytest.raises(CLIError, match="File not found|failed to read"): + load_tc_params_mapping(missing_file, ["TC-ACE-1.1"]) + + def test_invalid_json_raises(self, tmp_path: Path) -> None: + """Broken JSON raises CLIError.""" + p = tmp_path / "bad.json" + p.write_text("{bad json", encoding="utf-8") + with pytest.raises(CLIError, match="Invalid JSON"): + load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + + def test_array_root_raises(self, tmp_path: Path) -> None: + """A JSON array at root raises CLIError.""" + p = tmp_path / "arr.json" + p.write_text("[]", encoding="utf-8") + with pytest.raises(CLIError, match="Expected a JSON object"): + load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + + def test_non_dict_entry_value_raises(self, tmp_path: Path) -> None: + """An entry whose value is not a dict raises CLIError.""" + p = _write_mapping(tmp_path, {"TC-ACE-1.1": "just-a-string"}) + with pytest.raises(CLIError, match="must be a JSON object"): + load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + + def test_null_entry_value_raises(self, tmp_path: Path) -> None: + """A null entry value raises CLIError (null is not a dict).""" + p = _write_mapping(tmp_path, {"TC-ACE-1.1": None}) + with pytest.raises(CLIError, match="must be a JSON object"): + load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + + # ------------------------------------------------------------------ + # Real-world format variations + # ------------------------------------------------------------------ + + def test_multiple_space_separated_values_in_int_arg(self, tmp_path: Path) -> None: + """Space-separated NAME:VALUE pairs in int-arg are passed through unchanged.""" + p = _write_mapping(tmp_path, { + "TC-AVSM-2.10": { + "int-arg": "PIXIT.AVSM.ENDPOINT:1 PIXIT.AVSM.STREAMTYPE:1", + "timeout": "600", + } + }) + params, missing = load_tc_params_mapping(str(p), ["TC-AVSM-2.10"]) + assert params["int-arg"] == "PIXIT.AVSM.ENDPOINT:1 PIXIT.AVSM.STREAMTYPE:1" + assert params["timeout"] == "600" + assert missing == [] + + def test_mapping_keys_with_mixed_separators(self, tmp_path: Path) -> None: + """Mapping file keys with mixed -/_ separators are all normalised correctly.""" + p = _write_mapping(tmp_path, { + "TC-MCORE_FS-1.3": {"string-arg": "PIXIT.MCORE.EP:3"}, + }) + # Caller uses the same mixed format + params, missing = load_tc_params_mapping(str(p), ["TC-MCORE_FS-1.3"]) + assert params == {"string-arg": "PIXIT.MCORE.EP:3"} + assert missing == [] + + def test_mapping_superset_only_requested_ids_merged(self, tmp_path: Path) -> None: + """Params for TC IDs not in test_ids are not included in merged_params.""" + p = _write_mapping(tmp_path, { + "TC-ACE-1.1": {"int-arg": "PIXIT.ACE.EP:1"}, + "TC-CC-1.1": {"timeout": "300"}, + "TC-ICT-1.1": {"string-arg": "PIXIT.ICT.NAME:foo"}, + }) + params, missing = load_tc_params_mapping(str(p), ["TC-ACE-1.1"]) + assert params == {"int-arg": "PIXIT.ACE.EP:1"} + assert "timeout" not in params + assert "string-arg" not in params diff --git a/th_cli/commands/run_tests.py b/th_cli/commands/run_tests.py index 6b7532a..42554ce 100644 --- a/th_cli/commands/run_tests.py +++ b/th_cli/commands/run_tests.py @@ -42,8 +42,8 @@ from th_cli.exceptions import CLIError, handle_api_error from th_cli.test_run.camera.two_way_talk_handler import TwoWayTalkHandler from th_cli.test_run.websocket import TestRunSocket -from th_cli.utils import DEFAULT_CLI_PROJECT_NAME, build_test_selection, convert_nested_to_dict, load_json_config, merge_configs, read_pics_config -from th_cli.validation import validate_directory_path, validate_file_path, validate_test_ids +from th_cli.utils import DEFAULT_CLI_PROJECT_NAME, build_test_selection, convert_nested_to_dict, load_json_config, load_tc_params_mapping, merge_configs, read_pics_config +from th_cli.validation import validate_directory_path, validate_file_path, validate_tc_params_file, validate_test_ids # Constants JSON_INDENT = 2 @@ -85,6 +85,23 @@ type=click.Path(file_okay=False, dir_okay=True), help=colorize_help("Directory containing PICS XML configuration files. If not provided, no PICS will be used."), ) +@click.option( + "--tc-params-file", + "-m", + type=click.Path(file_okay=True, dir_okay=False), + help=colorize_help( + "Path to a JSON file that maps TC IDs to their test_parameters " + "(e.g. int-arg, string-arg, timeout). Entries are matched " + "case-insensitively with separators - _ . normalised. " + "TC IDs not found in the file are silently skipped." + "\n\n\b\nNOTE — Configuration precedence (lowest → highest priority):\n" + " 1. Project config (persistent, from TH project)\n" + " 2. --tc-params-file ← this option\n" + " 3. --config file (execution-only override)\n" + " 4. -- inline args (e.g. -- --int-arg PIXIT.X:1)\n" + "\nHigher-priority sources always win on conflicting keys." + ), +) @click.option( "--project-id", type=int, @@ -110,6 +127,7 @@ async def run_tests( tests_list: str, config: str | None = None, pics_config_folder: str | None = None, + tc_params_file: str | None = None, project_id: int | None = None, no_color: bool = False, no_streaming: bool = False, @@ -122,6 +140,7 @@ async def run_tests( tests_list: Comma-separated list of test case identifiers config: Optional path to JSON configuration file pics_config_folder: Optional path to directory containing PICS XML files + tc_params_file: Optional path to TC parameters mapping JSON file project_id: Optional project ID for the test run no_color: Flag to disable colored output @@ -146,6 +165,10 @@ async def run_tests( pics_path = validate_directory_path(pics_config_folder, must_exist=True) pics_config_folder = str(pics_path) + if tc_params_file: + tc_params_path = validate_tc_params_file(tc_params_file) + tc_params_file = str(tc_params_path) + client = None _webrtc_handler = None try: @@ -185,6 +208,34 @@ async def run_tests( execution_pics = await _get_project_pics(cli_project) click.echo(colorize_key_value("PICS Used (From Project)", json.dumps(execution_pics, indent=JSON_INDENT))) + # Auto-populate test_parameters from a TC params mapping file. + # Priority: project config < mapping file < --config override < -- inline args + if tc_params_file: + mapping_params, missing_ids = load_tc_params_mapping(tc_params_file, validated_test_ids) + if missing_ids: + click.echo( + colorize_warning( + f"TC params mapping: no entry found for " + f"{', '.join(missing_ids)} — " + "using existing config or defaults for those tests" + ) + ) + if mapping_params: + click.echo( + colorize_key_value( + "TC Params from Mapping File (Execution Only)", + json.dumps(mapping_params, indent=JSON_INDENT), + ) + ) + if "test_parameters" not in test_run_config or test_run_config["test_parameters"] is None: + test_run_config["test_parameters"] = {} + # Mapping params are merged first; --config values already in + # test_run_config take precedence (they were applied above). + test_run_config["test_parameters"] = { + **mapping_params, + **test_run_config["test_parameters"], + } + # Merge extra test parameters if provided (temporary for this execution only) if extra_test_params: click.echo( diff --git a/th_cli/utils.py b/th_cli/utils.py index fde074c..e2068a9 100644 --- a/th_cli/utils.py +++ b/th_cli/utils.py @@ -29,6 +29,7 @@ from th_cli.colorize import colorize_dump from th_cli.config import find_git_root, get_package_root from th_cli.exceptions import CLIError, handle_file_error +from th_cli.validation import parse_and_validate_tc_params_file # Constants DEFAULT_FILE_ENCODING = "utf-8" @@ -374,6 +375,76 @@ def read_pics_config(pics_config_folder: str) -> dict: return pics +def load_tc_params_mapping( + mapping_path: str, + test_ids: list[str], +) -> tuple[dict[str, Any], list[str]]: + """Load test_parameters for the given TC IDs from a mapping file. + + The mapping file is a JSON object whose keys are TC IDs and whose values + are ``test_parameters`` dicts with the same shape accepted by the backend + (e.g. ``int-arg``, ``string-arg``, ``timeout``). Both ``-``, ``_`` and + ``.`` separators are normalised, and key comparison is case-insensitive, + so any of the following spellings refer to the same test case:: + + "TC-ACE-1.1" "TC_ACE_1_1" "tc-ace-1.1" + + Example mapping file:: + + { + "TC-ACE-1.1": { "int-arg": "PIXIT.ACE.APPENDPOINT:1", "timeout": "300" }, + "TC-AVSM-2.10": { "int-arg": "PIXIT.AVSM.ENDPOINT:1 PIXIT.AVSM.STREAMTYPE:1" }, + "TC-MCORE-FS-1.3": { "string-arg": "PIXIT.MCORE.ENDPOINT:3" } + } + + All matched entries are merged into a single flat ``test_parameters`` dict. + When two entries share the same parameter key, the last matched entry wins + (order is alphabetical by normalised TC ID). + + Args: + mapping_path: Path to the JSON mapping file. + test_ids: List of TC IDs to look up (as returned by + ``validate_test_ids``). + + Returns: + A tuple of: + - ``merged_params`` – dict ready to be merged into + ``test_run_config["test_parameters"]``. + - ``missing_ids`` – list of TC IDs from *test_ids* that had no entry + in the mapping file (may be empty). + + Raises: + CLIError: If the file cannot be read, contains invalid JSON, or is not + a JSON object whose values are dicts. + """ + + def _normalise(tc_id: str) -> str: + return tc_id.replace("-", "_").replace(".", "_").upper() + + # Delegate file I/O and structural validation to the shared helper so the + # file is never parsed twice (validate_tc_params_file already calls it + # during pre-flight) and validation logic lives in one place. + raw = parse_and_validate_tc_params_file(mapping_path) + + # Build a normalised-key → params lookup. + normalised_mapping: dict[str, dict[str, Any]] = { + _normalise(key): value for key, value in raw.items() + } + + # Match each requested test ID against the normalised mapping. + merged_params: dict[str, Any] = {} + missing_ids: list[str] = [] + + for tc_id in sorted(test_ids): # sorted for deterministic merge order + norm = _normalise(tc_id) + if norm in normalised_mapping: + merged_params.update(normalised_mapping[norm]) + else: + missing_ids.append(tc_id) + + return merged_params, missing_ids + + def get_cli_version() -> str: """Get CLI version from pyproject.toml""" try: diff --git a/th_cli/validation.py b/th_cli/validation.py index 2550cd5..51cf86c 100644 --- a/th_cli/validation.py +++ b/th_cli/validation.py @@ -15,8 +15,10 @@ # """Input validation utilities for the CLI.""" +import json import re from pathlib import Path +from typing import Any from th_cli.exceptions import CLIError @@ -111,6 +113,83 @@ def validate_test_ids(test_ids: str) -> list[str]: return ids +def parse_and_validate_tc_params_file(file_path: str) -> dict[str, dict[str, Any]]: + """Read, parse, and validate a TC params mapping file, returning its contents. + + This is the single source of truth for mapping-file I/O and structural + validation. Both :func:`validate_tc_params_file` (pre-flight check) and + ``load_tc_params_mapping`` (runtime loader) delegate here so the file is + never parsed twice and validation logic lives in one place. + + Args: + file_path: Path to the JSON mapping file. + + Returns: + The parsed mapping as a ``dict[str, dict[str, Any]]`` — TC ID keys + mapping to their parameter dicts. + + Raises: + CLIError: If the file is missing, not a regular file, contains + invalid JSON, or does not match the expected top-level structure. + """ + path = validate_file_path(file_path, must_exist=True) + + try: + with open(path, "r", encoding="utf-8") as f: + data: Any = json.load(f) + except json.JSONDecodeError as e: + raise CLIError( + f"Invalid JSON in TC params mapping file '{file_path}': " + f"{e.msg} (line {e.lineno}, column {e.colno})" + ) + except OSError as e: + raise CLIError(f"Failed to read TC params mapping file '{file_path}': {e}") + + if not isinstance(data, dict): + raise CLIError( + f"Invalid TC params mapping file '{file_path}': " + f"Expected a JSON object at the top level, got {type(data).__name__}" + ) + + bad_entries = [k for k, v in data.items() if not isinstance(v, dict)] + if bad_entries: + raise CLIError( + f"Invalid TC params mapping file '{file_path}': " + f"Each entry value must be a JSON object (dict of parameters). " + f"Non-dict values found for TC IDs: {', '.join(bad_entries)}" + ) + + return data + + +def validate_tc_params_file(file_path: str) -> Path: + """Validate that a TC params mapping file exists and has the expected structure. + + Performs two levels of checking: + + 1. **File-level**: path exists, is a regular file, and is readable. + 2. **Format-level**: the file contains valid JSON whose top-level value is + a JSON object (dict), and whose values are themselves JSON objects. + + This is a *fast pre-flight* check intended to surface obvious problems + before the full run starts. The actual parsing is delegated to + :func:`parse_and_validate_tc_params_file` so the logic is not duplicated + with ``load_tc_params_mapping``. + + Args: + file_path: Path to the JSON mapping file. + + Returns: + Resolved ``Path`` object for the validated file. + + Raises: + CLIError: If the file is missing, not a regular file, contains + invalid JSON, or does not match the expected top-level structure. + """ + parse_and_validate_tc_params_file(file_path) + return Path(file_path).resolve() + + def validate_hostname(hostname: str) -> str: """Validate hostname format.""" if not hostname or len(hostname.strip()) == 0: