diff --git a/README.md b/README.md index fb7c9b1..2b56a8c 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,16 @@ To see available commands for an alias: xts myalias --help ``` +### Validate an .xts File + +Use the built-in validator to check YAML syntax for an `.xts` file without running it: + +```sh +xts validate /path/to/file.xts +``` + +This command reports syntax errors clearly and exits with code `0` for valid files or `1` for invalid files. + ## Example .xts File ```yaml diff --git a/src/xts_core/plugins/xts_allocator_client.py b/src/xts_core/plugins/xts_allocator_client.py index 07c6d55..e6099b9 100755 --- a/src/xts_core/plugins/xts_allocator_client.py +++ b/src/xts_core/plugins/xts_allocator_client.py @@ -76,6 +76,11 @@ def _send_request(method, url, data=None) -> dict: except requests.exceptions.RequestException as e: plugin_utils.error(f'Error during request: {e}') + @staticmethod + def send_request(method, url, data=None) -> dict: + """Compatibility wrapper expected by tests: delegates to _send_request.""" + return XTSAllocatorClient._send_request(method, url, data) + @staticmethod def _format_slots_list_to_table(response:list[dict]) -> Table: diff --git a/src/xts_core/xts.py b/src/xts_core/xts.py index 910e82e..fd2dc74 100755 --- a/src/xts_core/xts.py +++ b/src/xts_core/xts.py @@ -36,6 +36,7 @@ import argparse import os import re +import shlex import sys import yaml @@ -145,6 +146,7 @@ def _get_yaml_command_choices(self) -> list[tuple]: def _get_command_sections(self) -> dict: """Extract command sections from loaded XTS configuration.""" command_sections = {} + self._ignored_sections = [] def _is_command_section(subdict: dict) -> bool: for key, value in subdict.items(): @@ -153,13 +155,16 @@ def _is_command_section(subdict: dict) -> bool: elif isinstance(value, dict) and _is_command_section(value): return True return False - + if not isinstance(self._xts_config, dict): return command_sections for key, value in self._xts_config.items(): - if isinstance(value, dict) and _is_command_section(value): - command_sections[key] = value + if isinstance(value, dict): + if _is_command_section(value): + command_sections[key] = value + else: + self._ignored_sections.append(key) return command_sections @@ -168,9 +173,8 @@ def _setup_first_parser(self): """ Parse CLI arguments and set up argparse for all commands. The first argument must be either: - - a built-in options (currently only "alias") + - a built-in options ("alias" or "validate") - an alias name (resolved via ~/.xts/aliases.json to an .xts file path) - Direct .xts file usage from cwd or as the first argument is not supported. Returns: list[str]: Remaining args starting with the command name, e.g. ["run", ...]. @@ -190,7 +194,105 @@ def _setup_first_parser(self): alias_parser = first_arg_subparsers.add_parser('alias', help='Manage aliases (add, list, remove)') xts_alias.setup_alias_parser(alias_parser) + + validate_parser = first_arg_subparsers.add_parser( + 'validate', + help='Validate an .xts file and report syntax issues', + add_help=False, + ) + validate_parser.add_argument('path', nargs='?', help='Path to the .xts file to validate') return first_arg_parser + + def _validate_command_value(self, value, path: str): + """Validate that a command entry is a string or a list of strings.""" + def _validate_shell_command(command: str, command_path: str): + try: + shlex.split(command, posix=True) + except ValueError as exc: + message = str(exc) + if 'closing quotation' in message.lower() or 'unmatched' in message.lower(): + raise ValueError(f'Invalid shell command at "{command_path}": unbalanced quotes') from exc + raise ValueError(f'Invalid shell command at "{command_path}": {message}') from exc + + if isinstance(value, str): + _validate_shell_command(value, path) + elif isinstance(value, list): + if not all(isinstance(item, str) for item in value): + raise ValueError(f'Invalid command list at "{path}": all entries must be strings') + for item in value: + _validate_shell_command(item, path) + else: + raise ValueError(f'Invalid command definition at "{path}": expected a string or list of strings') + + def _validate_xts_structure(self, node, path: str = 'root'): + """Validate the expected .xts structure recursively.""" + if isinstance(node, dict): + for key, value in node.items(): + node_path = f'{path}/{key}' + if key == 'command': + self._validate_command_value(value, node_path) + elif isinstance(value, list): + raise ValueError( + f'Invalid .xts structure at "{node_path}": lists are not supported ' + 'in xts command sections' + ) + elif isinstance(value, dict): + self._validate_xts_structure(value, node_path) + elif isinstance(node, list): + raise ValueError( + f'Invalid .xts structure at "{path}": root-level lists are not supported ' + 'in xts configuration' + ) + + def _run_validate_command(self, argv: list[str]): + """ + Validate an .xts file path provided in argv. Exits with code 0 on success + and 1 on any error. Prints brief messages to stdout. + """ + validate_help_parser = XTSArgumentParser( + prog='xts validate', + description='Validate an .xts file and report syntax issues', + ) + validate_help_parser.add_argument( + 'path', + nargs='?', + help='Path to the .xts file to validate', + ) + + args = validate_help_parser.parse_args(argv) + if not args.path: + validate_help_parser.print_help() + print('Example: xts validate examples/example.xts') + raise SystemExit(1) + + path = args.path + if not os.path.exists(path): + print('xts config specified does not exist') + raise SystemExit(1) + try: + with open(path, 'r', encoding='utf-8') as stream: + data = yaml.load(stream, SafeLoader) + except (yaml.scanner.ScannerError, yaml.parser.ParserError, yaml.YAMLError): + print('The xts file is incorrectly formatted: {}'.format(path)) + raise SystemExit(1) + + try: + if not isinstance(data, dict): + raise ValueError('Invalid xts structure: root must be a mapping') + self._xts_config = data + self._command_sections = self._get_command_sections() + self._validate_xts_structure(data) + if self._ignored_sections: + for ignored in self._ignored_sections: + print(f'Warning: section "{ignored}" will be ignored because it contains no command key') + if not self._command_sections: + print(f'No command sections found in xts file: {path}') + raise SystemExit(1) + print(f'Validation passed for: {path}') + raise SystemExit(0) + except ValueError as exc: + print(str(exc)) + raise SystemExit(1) from exc def _run_yaml_runner(self, alias:str, arguments:list[str]): resolved_xts_path = xts_alias.resolve_alias_to_xts_path(alias) @@ -249,6 +351,8 @@ def run(self): alias_name_subparser = list(filter(lambda x: x.dest == 'alias_name',parser._actions))[0] alias_subparser = alias_name_subparser.choices.get('alias') raise SystemExit(xts_alias.run_alias_builtin(alias_subparser)) + case 'validate': + self._run_validate_command(remaining_args if remaining_args else [args.get('path', '')]) case None|'alias_name': parser.print_help() raise SystemExit(0) diff --git a/test/test_xts_all_cases.py b/test/test_xts_all_cases.py index 7fbcd3b..07e6ab8 100644 --- a/test/test_xts_all_cases.py +++ b/test/test_xts_all_cases.py @@ -6,6 +6,7 @@ sys.path.append(os.path.join(os.path.dirname(__file__), '../')) +from xts_core.xts import XTS from xts_core.xts_alias import ( add_alias, remove_alias, @@ -102,6 +103,98 @@ def test_multiple_aliases(mock_alias_config, tmp_path): aliases = load_aliases() assert "alias1" in aliases and "alias2" in aliases + +def test_validate_xts_file_success(tmp_path): + valid_file = tmp_path / "valid.xts" + valid_file.write_text( + "run:\n" + " hello_world:\n" + " command: echo \"hello world\"\n" + ) + + with pytest.raises(SystemExit) as excinfo: + XTS()._run_validate_command([str(valid_file)]) + + assert excinfo.value.code == 0 + + +def test_validate_cli_subcommand(monkeypatch, tmp_path): + valid_file = tmp_path / "valid.xts" + valid_file.write_text( + "run:\n" + " hello_world:\n" + " command: echo \"hello world\"\n" + ) + + monkeypatch.setattr(sys, "argv", ["xts", "validate", str(valid_file)]) + with pytest.raises(SystemExit) as excinfo: + XTS().run() + + assert excinfo.value.code == 0 + + +def test_validate_without_path_shows_usage(monkeypatch): + monkeypatch.setattr(sys, "argv", ["xts", "validate"]) + with patch('sys.stdout', new=StringIO()) as mock_stdout: + with pytest.raises(SystemExit) as excinfo: + XTS().run() + + assert excinfo.value.code == 1 + output = mock_stdout.getvalue().lower() + assert "usage: xts validate" in output + assert "example:" in output + + +def test_validate_xts_file_syntax_error(tmp_path): + invalid_file = tmp_path / "invalid.xts" + invalid_file.write_text("not: [valid: yaml") + + with patch('sys.stdout', new=StringIO()) as mock_stdout: + with pytest.raises(SystemExit) as excinfo: + XTS()._run_validate_command([str(invalid_file)]) + + assert excinfo.value.code == 1 + assert "incorrectly formatted" in mock_stdout.getvalue().lower() + + +def test_validate_xts_file_missing(tmp_path): + missing_file = tmp_path / "missing.xts" + with patch('sys.stdout', new=StringIO()) as mock_stdout: + with pytest.raises(SystemExit) as excinfo: + XTS()._run_validate_command([str(missing_file)]) + + assert excinfo.value.code == 1 + assert "does not exist" in mock_stdout.getvalue().lower() + + +def test_validate_xts_file_invalid_structure(tmp_path): + invalid_file = tmp_path / "invalid.xts" + invalid_file.write_text("run:\n - name: hello\n command: echo \"hello\"") + + with patch('sys.stdout', new=StringIO()) as mock_stdout: + with pytest.raises(SystemExit) as excinfo: + XTS()._run_validate_command([str(invalid_file)]) + + assert excinfo.value.code == 1 + assert "lists are not supported" in mock_stdout.getvalue().lower() + + +def test_validate_xts_file_invalid_command_string(tmp_path): + invalid_file = tmp_path / "invalid_command.xts" + invalid_file.write_text( + "run:\n" + " hello_world:\n" + " command: echo \"hello\n" + ) + + with patch('sys.stdout', new=StringIO()) as mock_stdout: + with pytest.raises(SystemExit) as excinfo: + XTS()._run_validate_command([str(invalid_file)]) + + assert excinfo.value.code == 1 + assert "unbalanced quotes" in mock_stdout.getvalue().lower() + + def test_allocator_add_slot_missing_args(monkeypatch): """Test add-slot with missing required arguments.""" client = XTSAllocatorClient()