From 234c80cf546d233851af9ee4e1e9fc444879c8e3 Mon Sep 17 00:00:00 2001 From: Rickard von Haugwitz Date: Sat, 20 Jun 2026 21:14:54 +0200 Subject: [PATCH] Add generated module front door handlers --- .../schemas/command_package_ir.schema.json | 316 ++++++++++++++++-- src/command_generation/targets/python.py | 241 ++++++++++++- tests/test_public_api.py | 297 +++++++++++++++- 3 files changed, 819 insertions(+), 35 deletions(-) diff --git a/src/command_generation/schemas/command_package_ir.schema.json b/src/command_generation/schemas/command_package_ir.schema.json index 7df8525..659181d 100644 --- a/src/command_generation/schemas/command_package_ir.schema.json +++ b/src/command_generation/schemas/command_package_ir.schema.json @@ -463,27 +463,306 @@ "runtime_module_handlers": { "type": "array", "items": { - "type": "object", - "required": [ - "operation_id", - "import_module", - "function" - ], - "properties": { - "operation_id": { - "type": "string", - "minLength": 1 + "oneOf": [ + { + "type": "object", + "required": [ + "operation_id", + "import_module", + "function" + ], + "properties": { + "operation_id": { + "type": "string", + "minLength": 1 + }, + "import_module": { + "type": "string", + "minLength": 1 + }, + "function": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + } + }, + "additionalProperties": false }, - "import_module": { - "type": "string", - "minLength": 1 + { + "type": "object", + "required": [ + "operation_id", + "handler", + "import_module", + "function" + ], + "properties": { + "operation_id": { + "type": "string", + "minLength": 1 + }, + "handler": { + "const": "argparse_function_call" + }, + "import_module": { + "type": "string", + "minLength": 1 + }, + "function": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "support_import_module": { + "type": "string", + "minLength": 1 + }, + "result": { + "enum": [ + "return_zero", + "return_int", + "emit_payload" + ] + }, + "emit_payload": { + "type": "object", + "properties": { + "import_module": { + "type": "string", + "minLength": 1 + }, + "function": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "format_attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + } + }, + "additionalProperties": false + }, + "arguments": { + "type": "array", + "items": { + "type": "object", + "required": [ + "name", + "kind" + ], + "properties": { + "name": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "kind": { + "enum": [ + "attr", + "bool_attr", + "list_attr", + "target_root", + "diagnostic_profile", + "module_descriptors" + ] + }, + "attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "default": {}, + "default_current": { + "type": "boolean" + }, + "allow_none": { + "type": "boolean" + }, + "validate_command": { + "type": "string", + "minLength": 1 + }, + "validate": { + "type": "boolean" + } + }, + "additionalProperties": false + } + } + }, + "additionalProperties": false }, - "function": { - "type": "string", - "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + { + "type": "object", + "required": [ + "operation_id", + "handler", + "command_attr", + "module_import", + "module_program", + "help_payload_import_module", + "help_payload_function", + "help_text_function", + "missing_module_message" + ], + "properties": { + "operation_id": { + "type": "string", + "minLength": 1 + }, + "handler": { + "const": "module_front_door" + }, + "command_attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "target_attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "format_attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "module_import": { + "type": "string", + "minLength": 1 + }, + "module_main": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "module_program": { + "type": "string", + "minLength": 1 + }, + "include_module_program": { + "type": "boolean" + }, + "help_payload_import_module": { + "type": "string", + "minLength": 1 + }, + "help_payload_function": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "help_text_import_module": { + "type": "string", + "minLength": 1 + }, + "help_text_function": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "missing_module_message": { + "type": "string", + "minLength": 1 + }, + "stdout_replacements": { + "type": "array", + "items": { + "type": "object", + "required": [ + "old", + "new" + ], + "properties": { + "old": { + "type": "string" + }, + "new": { + "type": "string" + } + }, + "additionalProperties": false + } + }, + "option_specs": { + "type": "array", + "items": { + "type": "object", + "required": [ + "option", + "attr" + ], + "properties": { + "option": { + "type": "string", + "minLength": 1 + }, + "attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "fallback_attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + }, + "kind": { + "enum": [ + "value", + "flag", + "repeated", + "repeated_group" + ] + } + }, + "additionalProperties": false + } + }, + "positionals": { + "type": "array", + "items": { + "type": "object", + "required": [ + "commands", + "attr" + ], + "properties": { + "commands": { + "type": "array", + "minItems": 1, + "items": { + "type": "string", + "minLength": 1 + } + }, + "attr": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + } + }, + "additionalProperties": false + } + }, + "local_command_handlers": { + "type": "array", + "items": { + "type": "object", + "required": [ + "command", + "import_module", + "function" + ], + "properties": { + "command": { + "type": "string", + "minLength": 1 + }, + "import_module": { + "type": "string", + "minLength": 1 + }, + "function": { + "type": "string", + "pattern": "^[A-Za-z_][A-Za-z0-9_]*$" + } + }, + "additionalProperties": false + } + } + }, + "additionalProperties": false } - }, - "additionalProperties": false + ] }, "description": "Operation handlers rendered into a generated runtime handler map by importing named runtime primitive functions." }, @@ -1513,4 +1792,3 @@ }, "x-command-generation-doc-role": "contract-reference" } - diff --git a/src/command_generation/targets/python.py b/src/command_generation/targets/python.py index f687a4c..8d69e36 100644 --- a/src/command_generation/targets/python.py +++ b/src/command_generation/targets/python.py @@ -126,15 +126,33 @@ def _python_command_module( } if operation_id in direct_handlers: handler = direct_handlers[operation_id] - import_module = str(handler["import_module"]) - imported_function = str(handler.get("function") or _runtime_adapter_function_name(operation_id)) - local_binding = _local_runtime_binding_for_import(package, import_module) - if local_binding is not None: - local_import = _command_module_import_for_binding(local_binding) - run_body = f" from {local_import} import {imported_function}\n\n return {imported_function}(args)\n" + if handler.get("handler") == "module_front_door": + runtime_module_file = _runtime_module_file_for_package(package) + if runtime_module_file == "cli": + rendered_handler = _render_module_front_door_runtime_handler("run", handler) + run_body = rendered_handler.split("def run(args: argparse.Namespace) -> int:\n", 1)[1] + support_imports = "import contextlib\nimport io\nimport json\nfrom ..cli import build_generated_parser\n" + else: + run_body = f" from ..{runtime_module_file} import _run_generated_operation\n\n return _run_generated_operation({operation_id!r}, args)\n" + support_imports = "" + elif handler.get("handler") == "argparse_function_call": + runtime_module_file = _runtime_module_file_for_package(package) + if runtime_module_file == "cli": + rendered_handler = _render_argparse_function_call_handler("run", handler) + run_body = rendered_handler.split("def run(args: argparse.Namespace) -> int:\n", 1)[1] + else: + run_body = f" from ..{runtime_module_file} import _run_generated_operation\n\n return _run_generated_operation({operation_id!r}, args)\n" + support_imports = "" else: - run_body = f" from {import_module} import {imported_function}\n\n return {imported_function}(args)\n" - support_imports = "" + import_module = str(handler["import_module"]) + imported_function = str(handler.get("function") or _runtime_adapter_function_name(operation_id)) + local_binding = _local_runtime_binding_for_import(package, import_module) + if local_binding is not None: + local_import = _command_module_import_for_binding(local_binding) + run_body = f" from {local_import} import {imported_function}\n\n return {imported_function}(args)\n" + else: + run_body = f" from {import_module} import {imported_function}\n\n return {imported_function}(args)\n" + support_imports = "" invoke_function = ( "\n\n" "def invoke(_values: Mapping[str, Any]) -> object:\n" @@ -650,6 +668,8 @@ def collect_handler_function(handler: dict[str, Any]) -> None: python_runtime_binding = package.get("python_runtime_binding", {}) if isinstance(python_runtime_binding, dict): for handler in python_runtime_binding.get("runtime_module_handlers", []): + if isinstance(handler, dict) and handler.get("handler") in {"module_front_door", "argparse_function_call"}: + continue if isinstance(handler, dict) and handler.get("import_module") == source_import_module: functions.add(str(handler.get("function") or _runtime_adapter_function_name(str(handler["operation_id"])))) return sorted(functions) @@ -1070,6 +1090,189 @@ def _runtime_adapter_function_name(operation_id: str) -> str: return "_run_" + "".join(character if character.isalnum() else "_" for character in operation_id) + "_adapter" +def _render_argparse_function_call_handler(function_name: str, handler: dict[str, Any]) -> str: + import_module = str(handler["import_module"]) + imported_function = str(handler["function"]) + support_import_module = str(handler.get("support_import_module") or import_module) + result_mode = str(handler.get("result", "return_zero")) + emit_payload = handler.get("emit_payload", {}) + emit_import_module = str(emit_payload.get("import_module") or support_import_module) if isinstance(emit_payload, dict) else "" + emit_function = str(emit_payload.get("function") or "_emit_payload") if isinstance(emit_payload, dict) else "_emit_payload" + emit_format_attr = str(emit_payload.get("format_attr") or "format") if isinstance(emit_payload, dict) else "format" + argument_specs = [spec for spec in handler.get("arguments", []) if isinstance(spec, dict)] + lines = [f"def {function_name}(args: argparse.Namespace) -> int:"] + kwargs: list[str] = [] + needs_support: set[str] = set() + for index, spec in enumerate(argument_specs): + kind = str(spec["kind"]) + name = str(spec["name"]) + variable_name = f"_arg_{index}_{name}" + if kind == "attr": + attr = str(spec["attr"]) + default = spec.get("default") + lines.append(f" {variable_name} = getattr(args, {attr!r}, {default!r})") + elif kind == "bool_attr": + attr = str(spec["attr"]) + lines.append(f" {variable_name} = bool(getattr(args, {attr!r}, False))") + elif kind == "list_attr": + attr = str(spec["attr"]) + lines.append(f" {variable_name} = list(getattr(args, {attr!r}, []) or [])") + elif kind == "target_root": + attr = str(spec.get("attr") or "target") + default_current = bool(spec.get("default_current", True)) + allow_none = bool(spec.get("allow_none", False)) + validate_command = str(spec.get("validate_command") or "") + needs_support.add("_resolve_target_root") + if validate_command: + needs_support.add("_validate_target_root") + if default_current: + lines.append(f" {variable_name} = _resolve_target_root(getattr(args, {attr!r}, None))") + else: + lines.append(f" {variable_name} = _resolve_target_root(getattr(args, {attr!r}, None)) if getattr(args, {attr!r}, None) else None") + if not allow_none: + lines.append(f" if {variable_name} is None:") + lines.append(" raise ValueError('target root resolution returned None')") + if validate_command: + if allow_none: + lines.append(f" if {variable_name} is not None:") + lines.append(f" _validate_target_root(command_name={validate_command!r}, target_root={variable_name})") + else: + lines.append(f" _validate_target_root(command_name={validate_command!r}, target_root={variable_name})") + elif kind == "diagnostic_profile": + default = str(spec.get("default") or "tiny") + needs_support.add("_diagnostic_profile") + lines.append(f" {variable_name} = _diagnostic_profile(args, default={default!r})") + elif kind == "module_descriptors": + needs_support.update({"_module_operations", "_validate_descriptor_contract"}) + lines.append(f" {variable_name} = _module_operations()") + if bool(spec.get("validate", True)): + lines.append(f" _validate_descriptor_contract({variable_name})") + else: + raise ValueError(f"unsupported argparse_function_call argument kind: {kind!r}") + kwargs.append(f"{name}={variable_name}") + if needs_support: + imported = ", ".join(sorted(needs_support)) + lines.insert(1, f" from {support_import_module} import {imported}") + lines.append(f" from {import_module} import {imported_function}") + call = f"{imported_function}({', '.join(kwargs)})" + if result_mode == "return_int": + lines.append(f" return int({call} or 0)") + elif result_mode == "emit_payload": + lines.append(f" payload = {call}") + lines.append(f" from {emit_import_module} import {emit_function}") + lines.append(f" {emit_function}(payload=payload, format_name=getattr(args, {emit_format_attr!r}, 'text'))") + lines.append(" return 0") + elif result_mode == "return_zero": + lines.append(f" {call}") + lines.append(" return 0") + else: + raise ValueError(f"unsupported argparse_function_call result mode: {result_mode!r}") + return "\n".join(lines) + "\n" + + +def _render_module_front_door_runtime_handler(function_name: str, handler: dict[str, Any]) -> str: + command_attr = str(handler["command_attr"]) + target_attr = str(handler.get("target_attr", "target")) + format_attr = str(handler.get("format_attr", "format")) + module_import = str(handler["module_import"]) + module_main = str(handler.get("module_main", "main")) + module_program = str(handler["module_program"]) + include_module_program = bool(handler.get("include_module_program", False)) + help_payload_import = str(handler["help_payload_import_module"]) + help_payload_function = str(handler["help_payload_function"]) + help_text_import = str(handler.get("help_text_import_module") or help_payload_import) + help_text_function = str(handler["help_text_function"]) + missing_message = str(handler["missing_module_message"]) + replacements = [ + (str(replacement["old"]), str(replacement["new"])) + for replacement in handler.get("stdout_replacements", []) + if isinstance(replacement, dict) + ] + option_specs = [ + { + "option": str(spec["option"]), + "attr": str(spec["attr"]), + "fallback_attr": str(spec.get("fallback_attr", "")), + "kind": str(spec.get("kind", "value")), + } + for spec in handler.get("option_specs", []) + if isinstance(spec, dict) + ] + positional_specs = [ + { + "commands": [str(command) for command in spec.get("commands", [])], + "attr": str(spec["attr"]), + } + for spec in handler.get("positionals", []) + if isinstance(spec, dict) + ] + local_handlers = { + str(local["command"]): (str(local["import_module"]), str(local["function"])) + for local in handler.get("local_command_handlers", []) + if isinstance(local, dict) + } + return ( + f"def {function_name}(args: argparse.Namespace) -> int:\n" + f" command_value = getattr(args, {command_attr!r}, None)\n" + " if not command_value:\n" + f" from {help_payload_import} import {help_payload_function} as help_payload_function\n\n" + f" payload = help_payload_function(target=getattr(args, {target_attr!r}, None))\n" + f" if getattr(args, {format_attr!r}, None) == 'json':\n" + " print(json.dumps(payload, indent=2))\n" + " else:\n" + f" from {help_text_import} import {help_text_function} as help_text_function\n\n" + " help_text_function(payload)\n" + " return 0\n" + f" local_handlers = {local_handlers!r}\n" + " local_handler = local_handlers.get(str(command_value))\n" + " if local_handler is not None:\n" + " module_name, function_name = local_handler\n" + " module = __import__(module_name, fromlist=[function_name])\n" + " return getattr(module, function_name)(args)\n" + f" argv = [{module_program!r}] if {include_module_program!r} else []\n" + " argv.append(str(command_value))\n" + f" for commands, attr in {[(spec['commands'], spec['attr']) for spec in positional_specs]!r}:\n" + " if str(command_value) in commands:\n" + " value = getattr(args, attr, None)\n" + " if value is not None and value != '' and value != []:\n" + " argv.append(str(value))\n" + f" for option, attr, fallback_attr, kind in {[(spec['option'], spec['attr'], spec['fallback_attr'], spec['kind']) for spec in option_specs]!r}:\n" + " value = getattr(args, attr, None)\n" + " if (value is None or value == [] or value == '') and fallback_attr:\n" + " value = getattr(args, fallback_attr, None)\n" + " if kind == 'flag':\n" + " if bool(value):\n" + " argv.append(option)\n" + " elif kind == 'repeated':\n" + " if isinstance(value, str):\n" + " value = [value]\n" + " for item in value or []:\n" + " argv.extend([option, str(item)])\n" + " elif kind == 'repeated_group':\n" + " if isinstance(value, str):\n" + " value = [value]\n" + " if value:\n" + " argv.append(option)\n" + " argv.extend(str(item) for item in value)\n" + " elif value is not None and value != '' and value != []:\n" + " argv.extend([option, str(value)])\n" + " try:\n" + f" module = __import__({module_import!r}, fromlist=[{module_main!r}])\n" + f" module_main = getattr(module, {module_main!r})\n" + " buffer = io.StringIO()\n" + " with contextlib.redirect_stdout(buffer):\n" + " result = module_main(argv)\n" + " output = buffer.getvalue()\n" + f" for old, new in {replacements!r}:\n" + " output = output.replace(old, new)\n" + " print(output, end='')\n" + " return int(result or 0)\n" + " except ImportError:\n" + f" build_generated_parser().error({missing_message!r})\n" + " return 2\n" + ) + + def _python_runtime_handler_module( package: dict[str, Any], binding: dict[str, Any], @@ -1090,13 +1293,18 @@ def _python_runtime_handler_module( function_name = _runtime_adapter_function_name(operation_id) if operation_id in direct_handlers: handler = direct_handlers[operation_id] - import_module = str(handler["import_module"]) - imported_function = str(handler.get("function") or function_name) - handler_functions.append( - f"def {function_name}(args: argparse.Namespace) -> int:\n" - f" from {import_module} import {imported_function}\n\n" - f" return {imported_function}(args)\n" - ) + if handler.get("handler") == "module_front_door": + handler_functions.append(_render_module_front_door_runtime_handler(function_name, handler)) + elif handler.get("handler") == "argparse_function_call": + handler_functions.append(_render_argparse_function_call_handler(function_name, handler)) + else: + import_module = str(handler["import_module"]) + imported_function = str(handler.get("function") or function_name) + handler_functions.append( + f"def {function_name}(args: argparse.Namespace) -> int:\n" + f" from {import_module} import {imported_function}\n\n" + f" return {imported_function}(args)\n" + ) else: handler_functions.append( f"def {function_name}(args: argparse.Namespace) -> int:\n" @@ -1111,6 +1319,9 @@ def _python_runtime_handler_module( '"""\n\n' "from __future__ import annotations\n\n" "import argparse\n" + "import contextlib\n" + "import io\n" + "import json\n" "import sys\n" "from typing import Any\n\n" "# DO NOT EDIT DIRECTLY.\n" diff --git a/tests/test_public_api.py b/tests/test_public_api.py index c4e35f0..74e1bec 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -50,7 +50,7 @@ from command_generation.conformance import TypescriptFunctionConformanceTarget, run_typescript_function_conformance_case from command_generation.primitive_executor import PrimitiveContext, execute_primitive from command_generation.targets.contract import PYTHON_TARGET_LAYOUT_VERSION, TYPESCRIPT_TARGET_LAYOUT_VERSION -from command_generation.targets.python import _python_local_runtime_binding_module +from command_generation.targets.python import _python_command_module, _python_local_runtime_binding_module, _python_runtime_handler_module def _maturity_policy() -> dict[str, object]: @@ -1338,6 +1338,301 @@ def second_value() -> str: sys.modules.pop(source_module.__name__, None) +def test_generated_module_front_door_handler_delegates_with_data_driven_argv_and_help() -> None: + runtime_module = types.ModuleType("fake_module_front_door_runtime") + calls: list[list[str]] = [] + + def module_main(argv: list[str]) -> int: + calls.append(argv) + print("demo-module route --target repo") + return 7 + + def help_payload(target: str | None = None) -> dict[str, object]: + return {"kind": "demo/help/v1", "target": target} + + def print_help(payload: dict[str, object]) -> None: + print(f"help:{payload['target']}") + + setattr(runtime_module, "module_main", module_main) + setattr(runtime_module, "help_payload", help_payload) + setattr(runtime_module, "print_help", print_help) + sys.modules[runtime_module.__name__] = runtime_module + try: + rendered = _python_runtime_handler_module( + { + "program": "demo-cli", + "python_runtime_binding": { + "operation_executor": { + "module_file": "primitives.operation_executor", + "supported_operation_ids": [], + } + }, + }, + { + "runtime_module_handlers": [ + { + "operation_id": "demo.front-door", + "handler": "module_front_door", + "command_attr": "demo_command", + "module_import": runtime_module.__name__, + "module_main": "module_main", + "module_program": "demo-module", + "help_payload_import_module": runtime_module.__name__, + "help_payload_function": "help_payload", + "help_text_function": "print_help", + "missing_module_message": "demo module is required", + "stdout_replacements": [{"old": "demo-module ", "new": "demo-cli demo "}], + "positionals": [{"commands": ["route"], "attr": "route_id"}], + "option_specs": [ + {"option": "--target", "attr": "target"}, + {"option": "--verbose", "attr": "verbose", "kind": "flag"}, + {"option": "--tag", "attr": "tags", "kind": "repeated"}, + {"option": "--group", "attr": "groups", "kind": "repeated_group"}, + {"option": "--path", "attr": "paths", "fallback_attr": "path", "kind": "repeated"}, + ], + } + ] + }, + source_path="demo_ir.json", + regenerate_command="generate-demo", + ) + assert "from fake_module_front_door_runtime import module_main" not in rendered + + class Parser: + def error(self, message: str) -> None: + raise AssertionError(message) + + generated_package = types.ModuleType("generated_demo") + setattr(generated_package, "build_generated_parser", lambda: Parser()) + setattr(generated_package, "generated_command_names", lambda: ["demo"]) + setattr(generated_package, "generated_operation_contract", lambda operation_id: {"id": operation_id}) + setattr(generated_package, "run_generated_command", lambda argv, handler: handler("demo.front-door", argv)) + setattr(generated_package, "supports_generated_command", lambda command: True) + primitives_package = types.ModuleType("generated_demo.primitives") + operation_executor_module = types.ModuleType("generated_demo.primitives.operation_executor") + setattr(operation_executor_module, "run_operation_ir", lambda operation, args: 0) + sys.modules["generated_demo"] = generated_package + sys.modules["generated_demo.primitives"] = primitives_package + sys.modules["generated_demo.primitives.operation_executor"] = operation_executor_module + module_globals: dict[str, object] = { + "__name__": "generated_demo.runtime", + "__package__": "generated_demo", + } + + exec(rendered, module_globals) + + args = types.SimpleNamespace(demo_command=None, target="repo", format="text") + assert cast(Callable[[str, object], int], module_globals["_run_generated_operation"])("demo.front-door", args) == 0 + + args = types.SimpleNamespace( + demo_command="route", + target="repo", + format="json", + verbose=True, + tags=["one", "two"], + groups=["alpha", "beta"], + route_id="R1", + paths=[], + path="fallback.txt", + ) + assert cast(Callable[[str, object], int], module_globals["_run_generated_operation"])("demo.front-door", args) == 7 + assert calls == [ + [ + "route", + "R1", + "--target", + "repo", + "--verbose", + "--tag", + "one", + "--tag", + "two", + "--group", + "alpha", + "beta", + "--path", + "fallback.txt", + ] + ] + finally: + sys.modules.pop(runtime_module.__name__, None) + sys.modules.pop("generated_demo", None) + sys.modules.pop("generated_demo.primitives", None) + sys.modules.pop("generated_demo.primitives.operation_executor", None) + + +def test_generated_module_front_door_command_uses_root_runtime_dispatcher() -> None: + rendered = _python_command_module( + { + "program": "demo-cli", + "python_runtime_binding": { + "runtime_module_file": "cli", + "operation_executor": { + "module_file": "primitives.operation_executor", + "supported_operation_ids": [], + }, + }, + }, + "demo.front-door", + { + "runtime_module_handlers": [ + { + "operation_id": "demo.front-door", + "handler": "module_front_door", + "command_attr": "demo_command", + "module_import": "demo_module.cli", + "module_program": "demo-module", + "help_payload_import_module": "demo_help", + "help_payload_function": "help_payload", + "help_text_function": "print_help", + "missing_module_message": "demo module is required", + } + ] + }, + source_path="demo_ir.json", + regenerate_command="generate-demo", + ) + + assert "from ..cli import build_generated_parser" in rendered + assert "command_value = getattr(args, 'demo_command', None)" in rendered + assert "_run_command_module" not in rendered + + +def test_generated_argparse_function_call_handler_maps_args_and_emits_payload() -> None: + runtime_module = types.ModuleType("fake_argparse_function_runtime") + calls: list[dict[str, object]] = [] + + def resolve_target_root(target: str | None) -> str: + return target or "repo" + + def validate_target_root(*, command_name: str, target_root: str) -> None: + calls.append({"validate": command_name, "target_root": target_root}) + + def diagnostic_profile(args: object, *, default: str) -> str: + return f"{default}:{getattr(args, 'profile', 'tiny')}" + + def payload_function( + *, + target_root: str, + changed_paths: list[str], + dry_run: bool, + task_text: str | None, + profile: str, + ) -> dict[str, object]: + payload = { + "target_root": target_root, + "changed_paths": changed_paths, + "dry_run": dry_run, + "task_text": task_text, + "profile": profile, + } + calls.append(payload) + return payload + + def emit_payload(*, payload: dict[str, object], format_name: str) -> None: + calls.append({"emit": format_name, "payload": payload}) + + setattr(runtime_module, "_resolve_target_root", resolve_target_root) + setattr(runtime_module, "_validate_target_root", validate_target_root) + setattr(runtime_module, "_diagnostic_profile", diagnostic_profile) + setattr(runtime_module, "payload_function", payload_function) + setattr(runtime_module, "_emit_payload", emit_payload) + sys.modules[runtime_module.__name__] = runtime_module + try: + rendered = _python_runtime_handler_module( + { + "program": "demo-cli", + "python_runtime_binding": { + "operation_executor": { + "module_file": "primitives.operation_executor", + "supported_operation_ids": [], + } + }, + }, + { + "runtime_module_handlers": [ + { + "operation_id": "demo.report", + "handler": "argparse_function_call", + "import_module": runtime_module.__name__, + "function": "payload_function", + "support_import_module": runtime_module.__name__, + "result": "emit_payload", + "emit_payload": {"import_module": runtime_module.__name__, "function": "_emit_payload"}, + "arguments": [ + { + "name": "target_root", + "kind": "target_root", + "attr": "target", + "validate_command": "demo", + }, + {"name": "changed_paths", "kind": "list_attr", "attr": "changed"}, + {"name": "dry_run", "kind": "bool_attr", "attr": "dry_run"}, + {"name": "task_text", "kind": "attr", "attr": "task"}, + {"name": "profile", "kind": "diagnostic_profile", "default": "tiny"}, + ], + } + ] + }, + source_path="demo_ir.json", + regenerate_command="generate-demo", + ) + + generated_package = types.ModuleType("generated_argparse_demo") + setattr(generated_package, "build_generated_parser", lambda: object()) + setattr(generated_package, "generated_command_names", lambda: ["demo"]) + setattr(generated_package, "generated_operation_contract", lambda operation_id: {"id": operation_id}) + setattr(generated_package, "run_generated_command", lambda argv, handler: handler("demo.report", argv)) + setattr(generated_package, "supports_generated_command", lambda command: True) + primitives_package = types.ModuleType("generated_argparse_demo.primitives") + operation_executor_module = types.ModuleType("generated_argparse_demo.primitives.operation_executor") + setattr(operation_executor_module, "run_operation_ir", lambda operation, args: 0) + sys.modules["generated_argparse_demo"] = generated_package + sys.modules["generated_argparse_demo.primitives"] = primitives_package + sys.modules["generated_argparse_demo.primitives.operation_executor"] = operation_executor_module + module_globals: dict[str, object] = { + "__name__": "generated_argparse_demo.runtime", + "__package__": "generated_argparse_demo", + } + + exec(rendered, module_globals) + args = types.SimpleNamespace( + target="repo", + changed=["a.py", "b.py"], + dry_run=True, + task="shape", + profile="full", + format="json", + ) + + assert cast(Callable[[str, object], int], module_globals["_run_generated_operation"])("demo.report", args) == 0 + assert calls == [ + {"validate": "demo", "target_root": "repo"}, + { + "target_root": "repo", + "changed_paths": ["a.py", "b.py"], + "dry_run": True, + "task_text": "shape", + "profile": "tiny:full", + }, + { + "emit": "json", + "payload": { + "target_root": "repo", + "changed_paths": ["a.py", "b.py"], + "dry_run": True, + "task_text": "shape", + "profile": "tiny:full", + }, + }, + ] + finally: + sys.modules.pop(runtime_module.__name__, None) + sys.modules.pop("generated_argparse_demo", None) + sys.modules.pop("generated_argparse_demo.primitives", None) + sys.modules.pop("generated_argparse_demo.primitives.operation_executor", None) + + def test_contract_owned_conformance_case_runs_black_box_cli(tmp_path: Path) -> None: contract = load_contract_conformance_case("todo.list.process") cli = tmp_path / "todo_cli.py"