Compare commits

...

2 Commits

11 changed files with 360 additions and 86 deletions

View File

@ -21,7 +21,8 @@ flx = Falyx("Deployment CLI")
flx.add_command( flx.add_command(
key="D", key="D",
aliases=["deploy"], aliases=["deploy"],
description="Deploy a service to a specified region.", description="Deploy",
help_text="Deploy a service to a specified region.",
action=Action( action=Action(
name="deploy_service", name="deploy_service",
action=deploy, action=deploy,
@ -31,6 +32,7 @@ flx.add_command(
"region": {"help": "Deployment region", "choices": ["us-east-1", "us-west-2"]}, "region": {"help": "Deployment region", "choices": ["us-east-1", "us-west-2"]},
"verbose": {"help": "Enable verbose mode"}, "verbose": {"help": "Enable verbose mode"},
}, },
tags=["deployment", "service"],
) )
deploy_chain = ChainedAction( deploy_chain = ChainedAction(
@ -48,8 +50,10 @@ deploy_chain = ChainedAction(
flx.add_command( flx.add_command(
key="N", key="N",
aliases=["notify"], aliases=["notify"],
description="Deploy a service and notify.", description="Deploy and Notify",
help_text="Deploy a service and notify.",
action=deploy_chain, action=deploy_chain,
tags=["deployment", "service", "notification"],
) )
asyncio.run(flx.run()) asyncio.run(flx.run())

View File

@ -128,13 +128,14 @@ class Command(BaseModel):
tags: list[str] = Field(default_factory=list) tags: list[str] = Field(default_factory=list)
logging_hooks: bool = False logging_hooks: bool = False
options_manager: OptionsManager = Field(default_factory=OptionsManager) options_manager: OptionsManager = Field(default_factory=OptionsManager)
arg_parser: CommandArgumentParser = Field(default_factory=CommandArgumentParser) arg_parser: CommandArgumentParser | None = None
arguments: list[dict[str, Any]] = Field(default_factory=list) arguments: list[dict[str, Any]] = Field(default_factory=list)
argument_config: Callable[[CommandArgumentParser], None] | None = None argument_config: Callable[[CommandArgumentParser], None] | None = None
custom_parser: ArgParserProtocol | None = None custom_parser: ArgParserProtocol | None = None
custom_help: Callable[[], str | None] | None = None custom_help: Callable[[], str | None] | None = None
auto_args: bool = True auto_args: bool = True
arg_metadata: dict[str, str | dict[str, Any]] = Field(default_factory=dict) arg_metadata: dict[str, str | dict[str, Any]] = Field(default_factory=dict)
simple_help_signature: bool = False
_context: ExecutionContext | None = PrivateAttr(default=None) _context: ExecutionContext | None = PrivateAttr(default=None)
@ -166,6 +167,12 @@ class Command(BaseModel):
raw_args, raw_args,
) )
return ((), {}) return ((), {})
if not isinstance(self.arg_parser, CommandArgumentParser):
logger.warning(
"[Command:%s] No argument parser configured, using default parsing.",
self.key,
)
return ((), {})
return await self.arg_parser.parse_args_split( return await self.arg_parser.parse_args_split(
raw_args, from_validate=from_validate raw_args, from_validate=from_validate
) )
@ -182,7 +189,9 @@ class Command(BaseModel):
def get_argument_definitions(self) -> list[dict[str, Any]]: def get_argument_definitions(self) -> list[dict[str, Any]]:
if self.arguments: if self.arguments:
return self.arguments return self.arguments
elif callable(self.argument_config): elif callable(self.argument_config) and isinstance(
self.arg_parser, CommandArgumentParser
):
self.argument_config(self.arg_parser) self.argument_config(self.arg_parser)
elif self.auto_args: elif self.auto_args:
if isinstance(self.action, BaseAction): if isinstance(self.action, BaseAction):
@ -218,8 +227,17 @@ class Command(BaseModel):
if self.logging_hooks and isinstance(self.action, BaseAction): if self.logging_hooks and isinstance(self.action, BaseAction):
register_debug_hooks(self.action.hooks) register_debug_hooks(self.action.hooks)
for arg_def in self.get_argument_definitions(): if self.arg_parser is None:
self.arg_parser.add_argument(*arg_def.pop("flags"), **arg_def) self.arg_parser = CommandArgumentParser(
command_key=self.key,
command_description=self.description,
command_style=self.style,
help_text=self.help_text,
help_epilogue=self.help_epilogue,
aliases=self.aliases,
)
for arg_def in self.get_argument_definitions():
self.arg_parser.add_argument(*arg_def.pop("flags"), **arg_def)
def _inject_options_manager(self) -> None: def _inject_options_manager(self) -> None:
"""Inject the options manager into the action if applicable.""" """Inject the options manager into the action if applicable."""
@ -317,6 +335,22 @@ class Command(BaseModel):
options_text = self.arg_parser.get_options_text(plain_text=True) options_text = self.arg_parser.get_options_text(plain_text=True)
return f" {command_keys_text:<20} {options_text} " return f" {command_keys_text:<20} {options_text} "
@property
def help_signature(self) -> str:
"""Generate a help signature for the command."""
if self.arg_parser and not self.simple_help_signature:
signature = [self.arg_parser.get_usage()]
signature.append(f" {self.help_text or self.description}")
if self.tags:
signature.append(f" [dim]Tags: {', '.join(self.tags)}[/dim]")
return "\n".join(signature).strip()
command_keys = " | ".join(
[f"[{self.style}]{self.key}[/{self.style}]"]
+ [f"[{self.style}]{alias}[/{self.style}]" for alias in self.aliases]
)
return f"{command_keys} {self.description}"
def log_summary(self) -> None: def log_summary(self) -> None:
if self._context: if self._context:
self._context.log_summary() self._context.log_summary()

View File

@ -118,14 +118,6 @@ def convert_commands(raw_commands: list[dict[str, Any]]) -> list[Command]:
commands = [] commands = []
for entry in raw_commands: for entry in raw_commands:
raw_command = RawCommand(**entry) raw_command = RawCommand(**entry)
parser = CommandArgumentParser(
command_key=raw_command.key,
command_description=raw_command.description,
command_style=raw_command.style,
help_text=raw_command.help_text,
help_epilogue=raw_command.help_epilogue,
aliases=raw_command.aliases,
)
commands.append( commands.append(
Command.model_validate( Command.model_validate(
{ {
@ -133,7 +125,6 @@ def convert_commands(raw_commands: list[dict[str, Any]]) -> list[Command]:
"action": wrap_if_needed( "action": wrap_if_needed(
import_action(raw_command.action), name=raw_command.description import_action(raw_command.action), name=raw_command.description
), ),
"arg_parser": parser,
} }
) )
) )

View File

@ -284,6 +284,7 @@ class Falyx:
action=Action("Exit", action=_noop), action=Action("Exit", action=_noop),
aliases=["EXIT", "QUIT"], aliases=["EXIT", "QUIT"],
style=OneColors.DARK_RED, style=OneColors.DARK_RED,
simple_help_signature=True,
) )
def _get_history_command(self) -> Command: def _get_history_command(self) -> Command:
@ -294,60 +295,70 @@ class Falyx:
aliases=["HISTORY"], aliases=["HISTORY"],
action=Action(name="View Execution History", action=er.summary), action=Action(name="View Execution History", action=er.summary),
style=OneColors.DARK_YELLOW, style=OneColors.DARK_YELLOW,
simple_help_signature=True,
) )
async def _show_help(self): async def _show_help(self, tag: str = "") -> None:
table = Table(title="[bold cyan]Help Menu[/]", box=box.SIMPLE) if tag:
table.add_column("Key", style="bold", no_wrap=True) table = Table(
table.add_column("Aliases", style="dim", no_wrap=True) title=tag.upper(),
table.add_column("Description", style="dim", overflow="fold") title_justify="left",
table.add_column("Tags", style="dim", no_wrap=True) show_header=False,
box=box.SIMPLE,
for command in self.commands.values(): show_footer=False,
help_text = command.help_text or command.description
table.add_row(
f"[{command.style}]{command.key}[/]",
", ".join(command.aliases) if command.aliases else "",
help_text,
", ".join(command.tags) if command.tags else "",
) )
tag_lower = tag.lower()
table.add_row( commands = [
f"[{self.exit_command.style}]{self.exit_command.key}[/]", command
", ".join(self.exit_command.aliases), for command in self.commands.values()
"Exit this menu or program", if any(tag_lower == tag.lower() for tag in command.tags)
) ]
for command in commands:
if self.history_command: table.add_row(command.help_signature)
table.add_row( self.console.print(table)
f"[{self.history_command.style}]{self.history_command.key}[/]", return
", ".join(self.history_command.aliases), else:
"History of executed actions", table = Table(
title="Help",
title_justify="left",
title_style=OneColors.LIGHT_YELLOW_b,
show_header=False,
show_footer=False,
box=box.SIMPLE,
) )
for command in self.commands.values():
table.add_row(command.help_signature)
if self.help_command: if self.help_command:
table.add_row( table.add_row(self.help_command.help_signature)
f"[{self.help_command.style}]{self.help_command.key}[/]", if self.history_command:
", ".join(self.help_command.aliases), table.add_row(self.history_command.help_signature)
"Show this help menu", table.add_row(self.exit_command.help_signature)
) table.add_row(f"Tip: '[{OneColors.LIGHT_YELLOW}]?[KEY][/]' to preview a command ")
self.console.print(table)
self.console.print(table, justify="center")
if self.mode == FalyxMode.MENU:
self.console.print(
f"📦 Tip: '[{OneColors.LIGHT_YELLOW}]?[KEY][/]' to preview a command "
"before running it.\n",
justify="center",
)
def _get_help_command(self) -> Command: def _get_help_command(self) -> Command:
"""Returns the help command for the menu.""" """Returns the help command for the menu."""
parser = CommandArgumentParser(
command_key="H",
command_description="Help",
command_style=OneColors.LIGHT_YELLOW,
aliases=["?", "HELP", "LIST"],
)
parser.add_argument(
"-t",
"--tag",
nargs="?",
default="",
help="Optional tag to filter commands by.",
)
return Command( return Command(
key="H", key="H",
aliases=["HELP", "?"], aliases=["?", "HELP", "LIST"],
description="Help", description="Help",
help_text="Show this help menu",
action=Action("Help", self._show_help), action=Action("Help", self._show_help),
style=OneColors.LIGHT_YELLOW, style=OneColors.LIGHT_YELLOW,
arg_parser=parser,
) )
def _get_completer(self) -> WordCompleter: def _get_completer(self) -> WordCompleter:
@ -568,7 +579,9 @@ class Falyx:
if not isinstance(submenu, Falyx): if not isinstance(submenu, Falyx):
raise NotAFalyxError("submenu must be an instance of Falyx.") raise NotAFalyxError("submenu must be an instance of Falyx.")
self._validate_command_key(key) self._validate_command_key(key)
self.add_command(key, description, submenu.menu, style=style) self.add_command(
key, description, submenu.menu, style=style, simple_help_signature=True
)
if submenu.exit_command.key == "X": if submenu.exit_command.key == "X":
submenu.update_exit_command(key="B", description="Back", aliases=["BACK"]) submenu.update_exit_command(key="B", description="Back", aliases=["BACK"])
@ -630,6 +643,7 @@ class Falyx:
custom_help: Callable[[], str | None] | None = None, custom_help: Callable[[], str | None] | None = None,
auto_args: bool = True, auto_args: bool = True,
arg_metadata: dict[str, str | dict[str, Any]] | None = None, arg_metadata: dict[str, str | dict[str, Any]] | None = None,
simple_help_signature: bool = False,
) -> Command: ) -> Command:
"""Adds an command to the menu, preventing duplicates.""" """Adds an command to the menu, preventing duplicates."""
self._validate_command_key(key) self._validate_command_key(key)
@ -640,15 +654,6 @@ class Falyx:
"arg_parser must be an instance of CommandArgumentParser." "arg_parser must be an instance of CommandArgumentParser."
) )
arg_parser = arg_parser arg_parser = arg_parser
else:
arg_parser = CommandArgumentParser(
command_key=key,
command_description=description,
command_style=style,
help_text=help_text,
help_epilogue=help_epilogue,
aliases=aliases,
)
command = Command( command = Command(
key=key, key=key,
@ -682,6 +687,7 @@ class Falyx:
custom_help=custom_help, custom_help=custom_help,
auto_args=auto_args, auto_args=auto_args,
arg_metadata=arg_metadata or {}, arg_metadata=arg_metadata or {},
simple_help_signature=simple_help_signature,
) )
if hooks: if hooks:
@ -706,16 +712,16 @@ class Falyx:
def get_bottom_row(self) -> list[str]: def get_bottom_row(self) -> list[str]:
"""Returns the bottom row of the table for displaying additional commands.""" """Returns the bottom row of the table for displaying additional commands."""
bottom_row = [] bottom_row = []
if self.history_command:
bottom_row.append(
f"[{self.history_command.key}] [{self.history_command.style}]"
f"{self.history_command.description}"
)
if self.help_command: if self.help_command:
bottom_row.append( bottom_row.append(
f"[{self.help_command.key}] [{self.help_command.style}]" f"[{self.help_command.key}] [{self.help_command.style}]"
f"{self.help_command.description}" f"{self.help_command.description}"
) )
if self.history_command:
bottom_row.append(
f"[{self.history_command.key}] [{self.history_command.style}]"
f"{self.history_command.description}"
)
bottom_row.append( bottom_row.append(
f"[{self.exit_command.key}] [{self.exit_command.style}]" f"[{self.exit_command.key}] [{self.exit_command.style}]"
f"{self.exit_command.description}" f"{self.exit_command.description}"
@ -727,12 +733,14 @@ class Falyx:
Build the standard table layout. Developers can subclass or call this Build the standard table layout. Developers can subclass or call this
in custom tables. in custom tables.
""" """
table = Table(title=self.title, show_header=False, box=box.SIMPLE, expand=True) # type: ignore[arg-type] table = Table(title=self.title, show_header=False, box=box.SIMPLE) # type: ignore[arg-type]
visible_commands = [item for item in self.commands.items() if not item[1].hidden] visible_commands = [item for item in self.commands.items() if not item[1].hidden]
space = self.console.width // self.columns
for chunk in chunks(visible_commands, self.columns): for chunk in chunks(visible_commands, self.columns):
row = [] row = []
for key, command in chunk: for key, command in chunk:
row.append(f"[{key}] [{command.style}]{command.description}") cell = f"[{key}] [{command.style}]{command.description}"
row.append(f"{cell:<{space}}")
table.add_row(*row) table.add_row(*row)
bottom_row = self.get_bottom_row() bottom_row = self.get_bottom_row()
for row in chunks(bottom_row, self.columns): for row in chunks(bottom_row, self.columns):
@ -1076,7 +1084,7 @@ class Falyx:
self.register_all_with_debug_hooks() self.register_all_with_debug_hooks()
if self.cli_args.command == "list": if self.cli_args.command == "list":
await self._show_help() await self._show_help(tag=self.cli_args.tag)
sys.exit(0) sys.exit(0)
if self.cli_args.command == "version" or self.cli_args.version: if self.cli_args.command == "version" or self.cli_args.version:

View File

@ -12,6 +12,7 @@ from rich.text import Text
from falyx.action.base import BaseAction from falyx.action.base import BaseAction
from falyx.exceptions import CommandArgumentError from falyx.exceptions import CommandArgumentError
from falyx.parsers.utils import coerce_value
from falyx.signals import HelpSignal from falyx.signals import HelpSignal
@ -290,7 +291,7 @@ class CommandArgumentParser:
for choice in choices: for choice in choices:
if not isinstance(choice, expected_type): if not isinstance(choice, expected_type):
try: try:
expected_type(choice) coerce_value(choice, expected_type)
except Exception: except Exception:
raise CommandArgumentError( raise CommandArgumentError(
f"Invalid choice {choice!r}: not coercible to {expected_type.__name__}" f"Invalid choice {choice!r}: not coercible to {expected_type.__name__}"
@ -303,7 +304,7 @@ class CommandArgumentParser:
"""Validate the default value type.""" """Validate the default value type."""
if default is not None and not isinstance(default, expected_type): if default is not None and not isinstance(default, expected_type):
try: try:
expected_type(default) coerce_value(default, expected_type)
except Exception: except Exception:
raise CommandArgumentError( raise CommandArgumentError(
f"Default value {default!r} for '{dest}' cannot be coerced to {expected_type.__name__}" f"Default value {default!r} for '{dest}' cannot be coerced to {expected_type.__name__}"
@ -316,7 +317,7 @@ class CommandArgumentParser:
for item in default: for item in default:
if not isinstance(item, expected_type): if not isinstance(item, expected_type):
try: try:
expected_type(item) coerce_value(item, expected_type)
except Exception: except Exception:
raise CommandArgumentError( raise CommandArgumentError(
f"Default list value {default!r} for '{dest}' cannot be coerced to {expected_type.__name__}" f"Default list value {default!r} for '{dest}' cannot be coerced to {expected_type.__name__}"
@ -595,7 +596,7 @@ class CommandArgumentParser:
i += new_i i += new_i
try: try:
typed = [spec.type(v) for v in values] typed = [coerce_value(value, spec.type) for value in values]
except Exception: except Exception:
raise CommandArgumentError( raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}" f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
@ -680,7 +681,9 @@ class CommandArgumentParser:
), "resolver should be an instance of BaseAction" ), "resolver should be an instance of BaseAction"
values, new_i = self._consume_nargs(args, i + 1, spec) values, new_i = self._consume_nargs(args, i + 1, spec)
try: try:
typed_values = [spec.type(value) for value in values] typed_values = [
coerce_value(value, spec.type) for value in values
]
except ValueError: except ValueError:
raise CommandArgumentError( raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}" f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
@ -709,7 +712,9 @@ class CommandArgumentParser:
assert result.get(spec.dest) is not None, "dest should not be None" assert result.get(spec.dest) is not None, "dest should not be None"
values, new_i = self._consume_nargs(args, i + 1, spec) values, new_i = self._consume_nargs(args, i + 1, spec)
try: try:
typed_values = [spec.type(value) for value in values] typed_values = [
coerce_value(value, spec.type) for value in values
]
except ValueError: except ValueError:
raise CommandArgumentError( raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}" f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
@ -724,7 +729,9 @@ class CommandArgumentParser:
assert result.get(spec.dest) is not None, "dest should not be None" assert result.get(spec.dest) is not None, "dest should not be None"
values, new_i = self._consume_nargs(args, i + 1, spec) values, new_i = self._consume_nargs(args, i + 1, spec)
try: try:
typed_values = [spec.type(value) for value in values] typed_values = [
coerce_value(value, spec.type) for value in values
]
except ValueError: except ValueError:
raise CommandArgumentError( raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}" f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"
@ -735,7 +742,9 @@ class CommandArgumentParser:
else: else:
values, new_i = self._consume_nargs(args, i + 1, spec) values, new_i = self._consume_nargs(args, i + 1, spec)
try: try:
typed_values = [spec.type(v) for v in values] typed_values = [
coerce_value(value, spec.type) for value in values
]
except ValueError: except ValueError:
raise CommandArgumentError( raise CommandArgumentError(
f"Invalid value for '{spec.dest}': expected {spec.type.__name__}" f"Invalid value for '{spec.dest}': expected {spec.type.__name__}"

View File

@ -255,6 +255,10 @@ def get_arg_parsers(
"list", help="List all available commands with tags" "list", help="List all available commands with tags"
) )
list_parser.add_argument(
"-t", "--tag", help="Filter commands by tag (case-insensitive)", default=None
)
version_parser = subparsers.add_parser("version", help="Show the Falyx version") version_parser = subparsers.add_parser("version", help="Show the Falyx version")
return FalyxParsers( return FalyxParsers(

View File

@ -24,7 +24,6 @@ def infer_args_from_func(
metadata = ( metadata = (
{"help": raw_metadata} if isinstance(raw_metadata, str) else raw_metadata {"help": raw_metadata} if isinstance(raw_metadata, str) else raw_metadata
) )
if param.kind not in ( if param.kind not in (
inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_ONLY,
inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.POSITIONAL_OR_KEYWORD,
@ -35,6 +34,8 @@ def infer_args_from_func(
arg_type = ( arg_type = (
param.annotation if param.annotation is not inspect.Parameter.empty else str param.annotation if param.annotation is not inspect.Parameter.empty else str
) )
if isinstance(arg_type, str):
arg_type = str
default = param.default if param.default is not inspect.Parameter.empty else None default = param.default if param.default is not inspect.Parameter.empty else None
is_required = param.default is inspect.Parameter.empty is_required = param.default is inspect.Parameter.empty
if is_required: if is_required:

View File

@ -1,10 +1,79 @@
from typing import Any import types
from datetime import datetime
from enum import EnumMeta
from typing import Any, Literal, Union, get_args, get_origin
from dateutil import parser as date_parser
from falyx.action.base import BaseAction from falyx.action.base import BaseAction
from falyx.logger import logger from falyx.logger import logger
from falyx.parsers.signature import infer_args_from_func from falyx.parsers.signature import infer_args_from_func
def coerce_bool(value: str) -> bool:
if isinstance(value, bool):
return value
value = value.strip().lower()
if value in {"true", "1", "yes", "on"}:
return True
elif value in {"false", "0", "no", "off"}:
return False
return bool(value)
def coerce_enum(value: Any, enum_type: EnumMeta) -> Any:
if isinstance(value, enum_type):
return value
if isinstance(value, str):
try:
return enum_type[value]
except KeyError:
pass
base_type = type(next(iter(enum_type)).value)
print(base_type)
try:
coerced_value = base_type(value)
return enum_type(coerced_value)
except (ValueError, TypeError):
raise ValueError(f"Value '{value}' could not be coerced to enum type {enum_type}")
def coerce_value(value: str, target_type: type) -> Any:
origin = get_origin(target_type)
args = get_args(target_type)
if origin is Literal:
if value not in args:
raise ValueError(
f"Value '{value}' is not a valid literal for type {target_type}"
)
return value
if isinstance(target_type, types.UnionType) or get_origin(target_type) is Union:
for arg in args:
try:
return coerce_value(value, arg)
except Exception:
continue
raise ValueError(f"Value '{value}' could not be coerced to any of {args!r}")
if isinstance(target_type, EnumMeta):
return coerce_enum(value, target_type)
if target_type is bool:
return coerce_bool(value)
if target_type is datetime:
try:
return date_parser.parse(value)
except ValueError as e:
raise ValueError(f"Value '{value}' could not be parsed as a datetime") from e
return target_type(value)
def same_argument_definitions( def same_argument_definitions(
actions: list[Any], actions: list[Any],
arg_metadata: dict[str, str | dict[str, Any]] | None = None, arg_metadata: dict[str, str | dict[str, Any]] | None = None,

View File

@ -1 +1 @@
__version__ = "0.1.43" __version__ = "0.1.45"

View File

@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "falyx" name = "falyx"
version = "0.1.43" version = "0.1.45"
description = "Reliable and introspectable async CLI action framework." description = "Reliable and introspectable async CLI action framework."
authors = ["Roland Thomas Jr <roland@rtj.dev>"] authors = ["Roland Thomas Jr <roland@rtj.dev>"]
license = "MIT" license = "MIT"
@ -16,6 +16,7 @@ python-json-logger = "^3.3.0"
toml = "^0.10" toml = "^0.10"
pyyaml = "^6.0" pyyaml = "^6.0"
aiohttp = "^3.11" aiohttp = "^3.11"
python-dateutil = "^2.8"
[tool.poetry.group.dev.dependencies] [tool.poetry.group.dev.dependencies]
pytest = "^8.3.5" pytest = "^8.3.5"

View File

@ -0,0 +1,153 @@
from datetime import datetime
from enum import Enum
from pathlib import Path
from typing import Literal
import pytest
from falyx.parsers.utils import coerce_value
# --- Tests ---
@pytest.mark.parametrize(
"value, target_type, expected",
[
("42", int, 42),
("3.14", float, 3.14),
("True", bool, True),
("hello", str, "hello"),
("", str, ""),
("False", bool, False),
],
)
def test_coerce_value_basic(value, target_type, expected):
assert coerce_value(value, target_type) == expected
@pytest.mark.parametrize(
"value, target_type, expected",
[
("42", int | float, 42),
("3.14", int | float, 3.14),
("hello", str | int, "hello"),
("1", bool | str, True),
],
)
def test_coerce_value_union_success(value, target_type, expected):
assert coerce_value(value, target_type) == expected
def test_coerce_value_union_failure():
with pytest.raises(ValueError) as excinfo:
coerce_value("abc", int | float)
assert "could not be coerced" in str(excinfo.value)
def test_coerce_value_typing_union_equivalent():
from typing import Union
assert coerce_value("123", Union[int, str]) == 123
assert coerce_value("abc", Union[int, str]) == "abc"
def test_coerce_value_edge_cases():
# int -> raises
with pytest.raises(ValueError):
coerce_value("not-an-int", int | float)
# empty string with str fallback
assert coerce_value("", int | str) == ""
# bool conversion
assert coerce_value("False", bool | str) is False
def test_coerce_value_enum():
class Color(Enum):
RED = "red"
GREEN = "green"
BLUE = "blue"
assert coerce_value("red", Color) == Color.RED
assert coerce_value("green", Color) == Color.GREEN
assert coerce_value("blue", Color) == Color.BLUE
with pytest.raises(ValueError):
coerce_value("yellow", Color) # Not a valid enum value
def test_coerce_value_int_enum():
class Status(Enum):
SUCCESS = 0
FAILURE = 1
PENDING = 2
assert coerce_value("0", Status) == Status.SUCCESS
assert coerce_value(1, Status) == Status.FAILURE
assert coerce_value("PENDING", Status) == Status.PENDING
assert coerce_value(Status.SUCCESS, Status) == Status.SUCCESS
with pytest.raises(ValueError):
coerce_value("3", Status)
with pytest.raises(ValueError):
coerce_value(3, Status)
class Mode(Enum):
DEV = "dev"
PROD = "prod"
def test_literal_coercion():
assert coerce_value("dev", Literal["dev", "prod"]) == "dev"
try:
coerce_value("staging", Literal["dev", "prod"])
assert False
except ValueError:
assert True
def test_enum_coercion():
assert coerce_value("dev", Mode) == Mode.DEV
assert coerce_value("DEV", Mode) == Mode.DEV
try:
coerce_value("staging", Mode)
assert False
except ValueError:
assert True
def test_union_coercion():
assert coerce_value("123", int | str) == 123
assert coerce_value("abc", int | str) == "abc"
assert coerce_value("False", bool | str) is False
def test_path_coercion():
result = coerce_value("/tmp/test.txt", Path)
assert isinstance(result, Path)
assert str(result) == "/tmp/test.txt"
def test_datetime_coercion():
result = coerce_value("2023-10-01T13:00:00", datetime)
assert isinstance(result, datetime)
assert result.year == 2023 and result.month == 10
with pytest.raises(ValueError):
coerce_value("not-a-date", datetime)
def test_bool_coercion():
assert coerce_value("true", bool) is True
assert coerce_value("False", bool) is False
assert coerce_value("0", bool) is False
assert coerce_value("", bool) is False
assert coerce_value("1", bool) is True
assert coerce_value("yes", bool) is True
assert coerce_value("no", bool) is False
assert coerce_value("on", bool) is True
assert coerce_value("off", bool) is False
assert coerce_value(True, bool) is True
assert coerce_value(False, bool) is False