mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-27 20:42:26 +00:00
* stash some changes * add more robust testing * mypy * point PR at V5 * introduce spec file * codespell * test fix * fix workflow environment setup * fix workflow environment setup * fix workflow environment setup * add pyyaml to dependencies * split lint jobs * fix workflow environment setup * fix workflow environment setup * workflow env setup * workflow env setup * clean up code comments * add auth hook entrypoints * codespell * add codegen feature * codespell * move _unpack into dispatchers for consistency with codegen packages * surface nested models in the response * fix missing coverage in CI * socrata updates * test fix * detect plotly output * add --include and --exclude flags from generate-extension command * cap test matrix at python 3.14 * no useless comments * platform controller command description split * merge URL overloads from path params * exclude none and unset from model dump --------- Co-authored-by: deeleeramone <> Co-authored-by: Copilot <copilot@github.com>
1163 lines
36 KiB
Python
1163 lines
36 KiB
Python
"""Test the Argparse Translator."""
|
|
|
|
from argparse import ArgumentParser
|
|
from datetime import date
|
|
from typing import Literal
|
|
|
|
import pytest
|
|
from pydantic import BaseModel
|
|
|
|
from openbb_cli.argparse_translator.argparse_argument import (
|
|
ArgparseArgumentGroupModel,
|
|
ArgparseArgumentModel,
|
|
)
|
|
from openbb_cli.argparse_translator.argparse_translator import (
|
|
ArgparseTranslator,
|
|
)
|
|
from openbb_cli.argparse_translator.reference_processor import (
|
|
ReferenceToArgumentsProcessor,
|
|
)
|
|
|
|
|
|
def sample_function(
|
|
symbol: str,
|
|
start_date: date | None = None,
|
|
limit: int = 100,
|
|
provider: Literal["fmp", "yfinance"] = "fmp",
|
|
) -> dict:
|
|
"""Sample function for testing."""
|
|
return {
|
|
"symbol": symbol,
|
|
"start_date": start_date,
|
|
"limit": limit,
|
|
"provider": provider,
|
|
}
|
|
|
|
|
|
def bool_function(
|
|
symbol: str,
|
|
adjusted: bool = False,
|
|
extended_hours: bool | None = None,
|
|
) -> dict:
|
|
"""Return dict with boolean parameters."""
|
|
return {"symbol": symbol, "adjusted": adjusted, "extended_hours": extended_hours}
|
|
|
|
|
|
def union_function(
|
|
param1: str | int = "default",
|
|
param2: str | int | None = None,
|
|
) -> dict:
|
|
"""Return dict with union type parameters."""
|
|
return {"param1": param1, "param2": param2}
|
|
|
|
|
|
def list_function(
|
|
symbols: list[str],
|
|
values: list[int] | None = None,
|
|
) -> dict:
|
|
"""Return dict with list parameters."""
|
|
return {"symbols": symbols, "values": values}
|
|
|
|
|
|
class CustomData(BaseModel):
|
|
"""Custom data model for testing."""
|
|
|
|
field1: str
|
|
field2: int = 10
|
|
|
|
|
|
def custom_type_function(data: CustomData) -> dict:
|
|
"""Return dict with custom type parameter."""
|
|
return {"data": data}
|
|
|
|
|
|
def test_get_action_type_for_bool_param():
|
|
"""Plain bool → store_true."""
|
|
|
|
def f(flag: bool = False) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
assert t._get_action_type(param) == "store_true"
|
|
|
|
|
|
def test_get_action_type_for_union_with_bool():
|
|
"""``bool | int`` (Union containing bool) → store_true."""
|
|
|
|
def f(flag: bool | int = False) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
assert t._get_action_type(param) == "store_true"
|
|
|
|
|
|
def test_get_action_type_for_optional_bool():
|
|
"""``bool | None`` → store_true (the Optional[bool] special case)."""
|
|
|
|
def f(flag: bool | None = None) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
assert t._get_action_type(param) == "store_true"
|
|
|
|
|
|
def test_get_type_and_choices_returns_str_for_str_or_int_union():
|
|
"""``str | int`` prefers str at the base type, with empty choices."""
|
|
|
|
def f(value: str | int = "x") -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
base_type, choices = t._get_type_and_choices(param)
|
|
assert base_type is str
|
|
|
|
|
|
def test_get_type_and_choices_strips_none_in_optional():
|
|
"""``str | None`` collapses to ``str``."""
|
|
|
|
def f(value: str | None = None) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
base_type, _ = t._get_type_and_choices(param)
|
|
assert base_type is str
|
|
|
|
|
|
def test_get_type_and_choices_extracts_literal_choices():
|
|
"""``Literal[a, b]`` produces choices ``(a, b)`` and base type matches."""
|
|
from typing import Literal
|
|
|
|
def f(provider: Literal["fmp", "yfinance"] = "fmp") -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
base_type, choices = t._get_type_and_choices(param)
|
|
assert base_type is str
|
|
assert set(choices) == {"fmp", "yfinance"}
|
|
|
|
|
|
def test_get_type_and_choices_handles_list_of_literal():
|
|
"""``list[Literal[...]]`` extracts inner choices."""
|
|
from typing import Literal
|
|
|
|
def f(items: list[Literal["a", "b"]] | None = None) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
_, choices = t._get_type_and_choices(param)
|
|
assert set(choices) == {"a", "b"}
|
|
|
|
|
|
def test_escape_help_doubles_percent_signs():
|
|
"""``_escape_help`` escapes lone ``%`` signs to satisfy argparse's %-formatting."""
|
|
assert ArgparseTranslator._escape_help("100% sure") == "100%% sure"
|
|
assert ArgparseTranslator._escape_help(None) is None
|
|
|
|
|
|
def test_param_is_default_detects_optional_params():
|
|
"""Helper distinguishes required vs defaulted parameters."""
|
|
import inspect
|
|
|
|
def f(a: int, b: int = 5) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
sig = inspect.signature(f)
|
|
params = list(sig.parameters.values())
|
|
assert ArgparseTranslator._param_is_default(params[0]) is False
|
|
assert ArgparseTranslator._param_is_default(params[1]) is True
|
|
|
|
|
|
def test_build_description_strips_section_headers():
|
|
"""``_build_description`` removes Parameters/Returns/Examples blocks from docstrings."""
|
|
docstring = (
|
|
"Get something.\n\n"
|
|
"Parameters\n"
|
|
"----------\n"
|
|
"x : int\n"
|
|
" The x value.\n\n"
|
|
"Returns\n"
|
|
"-------\n"
|
|
"int\n"
|
|
" The result.\n"
|
|
)
|
|
result = ArgparseTranslator._build_description(docstring)
|
|
assert "Parameters" not in result
|
|
assert "Returns" not in result
|
|
assert "Get something." in result
|
|
|
|
|
|
def test_build_description_empty_returns_empty():
|
|
"""Empty docstring → empty description."""
|
|
assert ArgparseTranslator._build_description("") == ""
|
|
|
|
|
|
def test_build_description_cleans_pipe_unions_in_param_lines():
|
|
"""Pipe-style annotations in body lines are humanized to 'or'."""
|
|
docstring = "Get something.\n\nx: int | str | None\n"
|
|
result = ArgparseTranslator._build_description(docstring)
|
|
assert "or" in result
|
|
|
|
|
|
def test_translator_handles_basemodel_param_flattens_fields():
|
|
"""BaseModel-typed params have their fields flattened into separate argparse arguments."""
|
|
t = ArgparseTranslator(custom_type_function)
|
|
parser = t.parser
|
|
optstrings = {opt for action in parser._actions for opt in action.option_strings}
|
|
assert any("data" in opt and "field1" in opt for opt in optstrings)
|
|
assert any("data" in opt and "field2" in opt for opt in optstrings)
|
|
|
|
|
|
def test_translator_required_vs_optional_param_routing():
|
|
"""Required params land in the 'required arguments' group; optional in standard."""
|
|
|
|
def f(req: str, opt: int = 5) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
parser = t.parser
|
|
groups = {
|
|
g.title: [a.dest for a in g._group_actions] for g in parser._action_groups
|
|
}
|
|
assert "req" in groups.get("required arguments", [])
|
|
optional_dest = []
|
|
for title, dests in groups.items():
|
|
if "option" in title.lower() and title != "required arguments":
|
|
optional_dest.extend(dests)
|
|
assert "opt" in optional_dest
|
|
|
|
|
|
def test_translator_skips_kwargs_param():
|
|
"""A ``**kwargs`` parameter is ignored (no argparse argument generated for it)."""
|
|
|
|
def f(name: str, **kwargs) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
parser = t.parser
|
|
optstrings = {opt for action in parser._actions for opt in action.option_strings}
|
|
assert not any("kwargs" in opt for opt in optstrings)
|
|
|
|
|
|
def test_translator_list_type_hint_uses_nargs_plus():
|
|
"""``list[int]`` parameters become ``nargs='+'`` arguments."""
|
|
|
|
t = ArgparseTranslator(list_function)
|
|
parser = t.parser
|
|
for action in parser._actions:
|
|
if action.dest == "symbols":
|
|
assert action.nargs == "+"
|
|
return
|
|
pytest.fail("symbols action not found")
|
|
|
|
|
|
def test_translator_bool_param_gets_store_true_action():
|
|
"""Bool params produce ``action='store_true'``."""
|
|
|
|
t = ArgparseTranslator(bool_function)
|
|
parser = t.parser
|
|
for action in parser._actions:
|
|
if action.dest == "adjusted":
|
|
from argparse import _StoreTrueAction
|
|
|
|
assert isinstance(action, _StoreTrueAction)
|
|
return
|
|
pytest.fail("adjusted action not found")
|
|
|
|
|
|
def test_translator_handle_argument_in_groups_merges_existing_argument():
|
|
"""When the same argument is added by two providers, choices are merged."""
|
|
|
|
def f(symbol: str, provider: Literal["fmp"] = "fmp") -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
extra_group = ArgparseArgumentGroupModel(
|
|
name="yfinance",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="symbol",
|
|
type=str,
|
|
dest="symbol",
|
|
default=None,
|
|
required=True,
|
|
action="store",
|
|
help=None,
|
|
nargs=None,
|
|
choices=("AAPL", "TSLA"),
|
|
)
|
|
],
|
|
)
|
|
t = ArgparseTranslator(f, custom_argument_groups=[extra_group])
|
|
parser = t.parser
|
|
for action in parser._actions:
|
|
if action.dest == "symbol":
|
|
choices = list(action.choices or [])
|
|
assert "AAPL" in choices or "TSLA" in choices
|
|
return
|
|
pytest.fail("symbol action not found")
|
|
|
|
|
|
def test_custom_argument_action_validation():
|
|
"""Test that CustomArgument raises an error for invalid actions."""
|
|
with pytest.raises(ValueError) as excinfo:
|
|
ArgparseArgumentModel(
|
|
name="test",
|
|
type=bool,
|
|
dest="test",
|
|
default=False,
|
|
required=True,
|
|
action="store",
|
|
help="Test argument",
|
|
nargs=None,
|
|
choices=None,
|
|
)
|
|
assert 'action must be "store_true"' in str(excinfo.value)
|
|
|
|
|
|
def test_custom_argument_remove_props_on_store_true():
|
|
"""Test that CustomArgument removes type, nargs, and choices on store_true."""
|
|
argument = ArgparseArgumentModel(
|
|
name="verbose",
|
|
type=None,
|
|
dest="verbose",
|
|
default=None,
|
|
required=False,
|
|
action="store_true",
|
|
help="Verbose output",
|
|
nargs=None,
|
|
choices=None,
|
|
)
|
|
assert argument.type is None
|
|
assert argument.nargs is None
|
|
assert argument.choices is None
|
|
|
|
|
|
def test_custom_argument_group():
|
|
"""Test the CustomArgumentGroup class."""
|
|
args = [
|
|
ArgparseArgumentModel(
|
|
name="test",
|
|
type=int,
|
|
dest="test",
|
|
default=1,
|
|
required=True,
|
|
action="store",
|
|
help="Test argument",
|
|
nargs=None,
|
|
choices=None,
|
|
)
|
|
]
|
|
group = ArgparseArgumentGroupModel(name="Test Group", arguments=args)
|
|
assert group.name == "Test Group"
|
|
assert len(group.arguments) == 1
|
|
assert group.arguments[0].name == "test"
|
|
|
|
|
|
def test_argparse_translator_setup():
|
|
"""Test the ArgparseTranslator setup."""
|
|
|
|
def test_function(test_arg: int):
|
|
"""A test function."""
|
|
return test_arg * 2
|
|
|
|
translator = ArgparseTranslator(func=test_function)
|
|
parser = translator.parser
|
|
assert isinstance(parser, ArgumentParser)
|
|
assert "--test_arg" in parser._option_string_actions
|
|
|
|
|
|
def test_argparse_translator_execution():
|
|
"""Test the ArgparseTranslator execution."""
|
|
|
|
def test_function(test_arg: int) -> int:
|
|
"""A test function."""
|
|
return test_arg * 2
|
|
|
|
translator = ArgparseTranslator(func=test_function)
|
|
parsed_args = translator.parser.parse_args(["--test_arg", "3"])
|
|
result = translator.execute_func(parsed_args)
|
|
assert result == 6
|
|
|
|
|
|
def test_basic_translation():
|
|
"""Test basic function translation."""
|
|
translator = ArgparseTranslator(sample_function)
|
|
parser = translator.parser
|
|
|
|
actions = {
|
|
action.dest: action for action in parser._actions if action.dest != "help"
|
|
}
|
|
|
|
assert "symbol" in actions
|
|
assert "start_date" in actions
|
|
assert "limit" in actions
|
|
assert "provider" in actions
|
|
|
|
|
|
def test_required_arguments():
|
|
"""Test that required arguments are correctly identified."""
|
|
translator = ArgparseTranslator(sample_function)
|
|
|
|
required_actions = [action for action in translator._required._group_actions]
|
|
assert any(action.dest == "symbol" for action in required_actions)
|
|
|
|
|
|
def test_optional_arguments():
|
|
"""Test that optional arguments have defaults."""
|
|
translator = ArgparseTranslator(sample_function)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
|
|
assert actions["limit"].default == 100
|
|
assert actions["provider"].default == "fmp"
|
|
|
|
|
|
def test_literal_choices():
|
|
"""Test that Literal types create choices."""
|
|
translator = ArgparseTranslator(sample_function)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
assert set(actions["provider"].choices) == {"fmp", "yfinance"}
|
|
|
|
|
|
def test_bool_store_true():
|
|
"""Test that bool parameters use store_true action."""
|
|
translator = ArgparseTranslator(bool_function)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
|
|
assert actions["adjusted"].__class__.__name__ == "_StoreTrueAction"
|
|
assert actions["extended_hours"].__class__.__name__ == "_StoreTrueAction"
|
|
|
|
|
|
def test_union_type_handling():
|
|
"""Test Union type handling."""
|
|
translator = ArgparseTranslator(union_function)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
|
|
assert actions["param1"].type is str
|
|
assert actions["param2"].type is str
|
|
|
|
|
|
def test_pipe_union_type_handling():
|
|
"""Test pipe union (|) type handling."""
|
|
translator = ArgparseTranslator(union_function)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
assert "param2" in actions
|
|
|
|
|
|
def test_list_nargs():
|
|
"""Test that list parameters get nargs='+'."""
|
|
translator = ArgparseTranslator(list_function)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
|
|
assert actions["symbols"].nargs == "+"
|
|
assert "values" in actions
|
|
|
|
|
|
def test_custom_type_flattening():
|
|
"""Test that custom BaseModel types are flattened."""
|
|
translator = ArgparseTranslator(custom_type_function)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
|
|
assert "data__field1" in actions
|
|
assert "data__field2" in actions
|
|
assert actions["data__field2"].default == 10
|
|
|
|
|
|
def test_unflatten_args():
|
|
"""Test unflattening of nested arguments."""
|
|
args_dict = {
|
|
"symbol": "AAPL",
|
|
"data__field1": "value1",
|
|
"data__field2": 20,
|
|
}
|
|
|
|
result = ArgparseTranslator._unflatten_args(args_dict)
|
|
|
|
assert result["symbol"] == "AAPL"
|
|
assert "data" in result
|
|
assert result["data"]["field1"] == "value1"
|
|
assert result["data"]["field2"] == 20
|
|
|
|
|
|
def test_description_cleaning_union():
|
|
"""Test that Union types are cleaned in descriptions."""
|
|
|
|
def func_with_union(param: str | int) -> None:
|
|
"""Test function.
|
|
|
|
Parameters
|
|
----------
|
|
param : Union[str, int]
|
|
A parameter description.
|
|
"""
|
|
|
|
translator = ArgparseTranslator(func_with_union)
|
|
description = translator._build_description(func_with_union.__doc__)
|
|
|
|
assert "Union[" not in description
|
|
|
|
|
|
def test_description_cleaning_optional():
|
|
"""Test that Optional types are cleaned in descriptions."""
|
|
|
|
def func_with_optional(param: str | None) -> None:
|
|
"""Test function.
|
|
|
|
Parameters
|
|
----------
|
|
param : Optional[str]
|
|
A parameter description.
|
|
"""
|
|
|
|
translator = ArgparseTranslator(func_with_optional)
|
|
description = translator._build_description(func_with_optional.__doc__)
|
|
|
|
assert "Optional[" not in description
|
|
|
|
|
|
def test_description_cleaning_annotated():
|
|
"""Test that Annotated types are cleaned in descriptions."""
|
|
|
|
def func_with_annotated(param: str) -> None:
|
|
"""Test function.
|
|
|
|
Parameters
|
|
----------
|
|
param : Annotated[str, Field(...)]
|
|
A parameter description.
|
|
"""
|
|
|
|
translator = ArgparseTranslator(func_with_annotated)
|
|
description = translator._build_description(func_with_annotated.__doc__)
|
|
|
|
assert "Annotated[" not in description
|
|
assert "Field" not in description
|
|
|
|
|
|
def test_description_cleaning_pipe_union():
|
|
"""Test that pipe unions (|) are cleaned in descriptions."""
|
|
|
|
def func_with_pipe(param: str | int | None) -> None:
|
|
"""Test function.
|
|
|
|
Parameters
|
|
----------
|
|
param : str | int | None
|
|
A parameter description.
|
|
"""
|
|
|
|
translator = ArgparseTranslator(func_with_pipe)
|
|
description = translator._build_description(func_with_pipe.__doc__)
|
|
|
|
assert " | " not in description
|
|
|
|
|
|
def test_custom_argument_groups():
|
|
"""Test adding custom argument groups."""
|
|
custom_groups = [
|
|
ArgparseArgumentGroupModel(
|
|
name="custom_provider",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="custom_param",
|
|
type=str,
|
|
dest="custom_param",
|
|
default="default",
|
|
required=False,
|
|
action="store",
|
|
help="Custom parameter",
|
|
nargs=None,
|
|
choices=("a", "b", "c"),
|
|
)
|
|
],
|
|
)
|
|
]
|
|
|
|
translator = ArgparseTranslator(
|
|
sample_function, custom_argument_groups=custom_groups
|
|
)
|
|
parser = translator.parser
|
|
|
|
actions = {action.dest: action for action in parser._actions}
|
|
assert "custom_param" in actions
|
|
assert set(actions["custom_param"].choices) == {"a", "b", "c"}
|
|
|
|
|
|
def test_provider_parameters_tracking():
|
|
"""Test that provider parameters are tracked."""
|
|
custom_groups = [
|
|
ArgparseArgumentGroupModel(
|
|
name="provider1",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="param1",
|
|
type=str,
|
|
dest="param1",
|
|
default=None,
|
|
required=False,
|
|
action="store",
|
|
help="Provider param",
|
|
nargs=None,
|
|
choices=None,
|
|
)
|
|
],
|
|
)
|
|
]
|
|
|
|
translator = ArgparseTranslator(
|
|
sample_function, custom_argument_groups=custom_groups
|
|
)
|
|
|
|
assert "provider1" in translator.provider_parameters
|
|
assert "param1" in translator.provider_parameters["provider1"]
|
|
|
|
|
|
class TestReferenceToArgumentsProcessor:
|
|
"""Test the ReferenceToArgumentsProcessor class."""
|
|
|
|
def test_parse_type_simple(self):
|
|
"""Test parsing simple types."""
|
|
assert ReferenceToArgumentsProcessor._parse_type("str") is str
|
|
assert ReferenceToArgumentsProcessor._parse_type("int") is int
|
|
assert ReferenceToArgumentsProcessor._parse_type("float") is float
|
|
assert ReferenceToArgumentsProcessor._parse_type("bool") is bool
|
|
|
|
def test_parse_type_optional(self):
|
|
"""Test parsing Optional types."""
|
|
assert ReferenceToArgumentsProcessor._parse_type("Optional[str]") is str
|
|
assert ReferenceToArgumentsProcessor._parse_type("Optional[int]") is int
|
|
assert ReferenceToArgumentsProcessor._parse_type("str | None") is str
|
|
assert ReferenceToArgumentsProcessor._parse_type("int | None") is int
|
|
|
|
def test_parse_type_literal(self):
|
|
"""Test parsing Literal types."""
|
|
assert ReferenceToArgumentsProcessor._parse_type("Literal['a', 'b']") is str
|
|
assert (
|
|
ReferenceToArgumentsProcessor._parse_type("Literal['option1', 'option2']")
|
|
is str
|
|
)
|
|
|
|
def test_parse_type_annotated(self):
|
|
"""Test parsing Annotated types."""
|
|
assert (
|
|
ReferenceToArgumentsProcessor._parse_type("Annotated[str, Field(...)]")
|
|
is str
|
|
)
|
|
assert (
|
|
ReferenceToArgumentsProcessor._parse_type("Annotated[int, Field(...)]")
|
|
is int
|
|
)
|
|
|
|
def test_parse_type_unknown(self):
|
|
"""Test parsing unknown types defaults to str."""
|
|
assert ReferenceToArgumentsProcessor._parse_type("UnknownType") is str
|
|
assert ReferenceToArgumentsProcessor._parse_type("CustomClass") is str
|
|
|
|
def test_get_nargs_list(self):
|
|
"""Test getting nargs for list types."""
|
|
processor = ReferenceToArgumentsProcessor({})
|
|
assert processor._get_nargs(list[str]) == "+"
|
|
assert processor._get_nargs(list[int]) == "+"
|
|
|
|
def test_get_nargs_non_list(self):
|
|
"""Test getting nargs for non-list types."""
|
|
processor = ReferenceToArgumentsProcessor({})
|
|
assert processor._get_nargs(str) is None
|
|
assert processor._get_nargs(int) is None
|
|
|
|
def test_get_choices_literal(self):
|
|
"""Test extracting choices from Literal types."""
|
|
processor = ReferenceToArgumentsProcessor({})
|
|
choices = processor._get_choices("Literal['a', 'b', 'c']", custom_choices=None)
|
|
assert set(choices) == {"a", "b", "c"}
|
|
|
|
def test_get_choices_multiple_literals(self):
|
|
"""Test extracting choices from multiple Literal types."""
|
|
processor = ReferenceToArgumentsProcessor({})
|
|
choices = processor._get_choices(
|
|
"Union[Literal['a', 'b'], Literal['c', 'd']]", custom_choices=None
|
|
)
|
|
assert set(choices) == {"a", "b", "c", "d"}
|
|
|
|
def test_get_choices_custom(self):
|
|
"""Test custom choices override Literal choices."""
|
|
processor = ReferenceToArgumentsProcessor({})
|
|
custom = ["x", "y", "z"]
|
|
choices = processor._get_choices("Literal['a', 'b']", custom_choices=custom)
|
|
assert choices == ("x", "y", "z")
|
|
|
|
def test_get_choices_none(self):
|
|
"""Test no choices when not Literal and no custom."""
|
|
processor = ReferenceToArgumentsProcessor({})
|
|
assert processor._get_choices("str", custom_choices=None) is None
|
|
assert processor._get_choices("int", custom_choices=None) is None
|
|
|
|
def test_build_custom_groups(self):
|
|
"""Test building custom argument groups from reference."""
|
|
reference = {
|
|
"/equity/price/historical": {
|
|
"parameters": {
|
|
"standard": [],
|
|
"fmp": [
|
|
{
|
|
"name": "interval",
|
|
"type": "Literal['1min', '5min', '15min']",
|
|
"description": "Time interval",
|
|
"default": "1min",
|
|
"optional": True,
|
|
"standard": False,
|
|
"choices": None,
|
|
}
|
|
],
|
|
}
|
|
}
|
|
}
|
|
|
|
processor = ReferenceToArgumentsProcessor(reference)
|
|
groups = processor.custom_groups
|
|
|
|
assert "/equity/price/historical" in groups
|
|
assert len(groups["/equity/price/historical"]) == 1
|
|
assert groups["/equity/price/historical"][0].name == "fmp"
|
|
assert len(groups["/equity/price/historical"][0].arguments) == 1
|
|
|
|
arg = groups["/equity/price/historical"][0].arguments[0]
|
|
assert arg.name == "interval"
|
|
assert arg.type is str
|
|
assert arg.default == "1min"
|
|
assert not arg.required
|
|
assert set(arg.choices) == {"1min", "5min", "15min"}
|
|
|
|
def test_build_custom_groups_skip_standard(self):
|
|
"""Test that standard parameters are skipped."""
|
|
reference = {
|
|
"/test/route": {
|
|
"parameters": {
|
|
"standard": [],
|
|
"provider1": [
|
|
{
|
|
"name": "param1",
|
|
"type": "str",
|
|
"description": "Standard param",
|
|
"default": None,
|
|
"optional": True,
|
|
"standard": True,
|
|
"choices": None,
|
|
},
|
|
{
|
|
"name": "param2",
|
|
"type": "int",
|
|
"description": "Custom param",
|
|
"default": 10,
|
|
"optional": False,
|
|
"standard": False,
|
|
"choices": None,
|
|
},
|
|
],
|
|
}
|
|
}
|
|
}
|
|
|
|
processor = ReferenceToArgumentsProcessor(reference)
|
|
groups = processor.custom_groups
|
|
|
|
assert len(groups["/test/route"][0].arguments) == 1
|
|
assert groups["/test/route"][0].arguments[0].name == "param2"
|
|
|
|
def test_build_custom_groups_multiple_providers(self):
|
|
"""Test building groups for multiple providers."""
|
|
reference = {
|
|
"/test/route": {
|
|
"parameters": {
|
|
"standard": [],
|
|
"provider1": [
|
|
{
|
|
"name": "param1",
|
|
"type": "str",
|
|
"description": "Provider 1 param",
|
|
"default": None,
|
|
"optional": True,
|
|
"standard": False,
|
|
"choices": None,
|
|
}
|
|
],
|
|
"provider2": [
|
|
{
|
|
"name": "param2",
|
|
"type": "int",
|
|
"description": "Provider 2 param",
|
|
"default": 5,
|
|
"optional": False,
|
|
"standard": False,
|
|
"choices": None,
|
|
}
|
|
],
|
|
}
|
|
}
|
|
}
|
|
|
|
processor = ReferenceToArgumentsProcessor(reference)
|
|
groups = processor.custom_groups
|
|
|
|
assert len(groups["/test/route"]) == 2
|
|
provider_names = {group.name for group in groups["/test/route"]}
|
|
assert provider_names == {"provider1", "provider2"}
|
|
|
|
def test_bool_action(self):
|
|
"""Test that bool types get store_true action."""
|
|
reference = {
|
|
"/test/route": {
|
|
"parameters": {
|
|
"standard": [],
|
|
"provider1": [
|
|
{
|
|
"name": "flag",
|
|
"type": "bool",
|
|
"description": "A boolean flag",
|
|
"default": False,
|
|
"optional": True,
|
|
"standard": False,
|
|
"choices": None,
|
|
}
|
|
],
|
|
}
|
|
}
|
|
}
|
|
|
|
processor = ReferenceToArgumentsProcessor(reference)
|
|
groups = processor.custom_groups
|
|
|
|
arg = groups["/test/route"][0].arguments[0]
|
|
assert arg.action == "store_true"
|
|
assert arg.type is None
|
|
|
|
|
|
def test_handle_argument_in_groups_required_arg_merges_choices():
|
|
"""A second provider supplying choices for a *required* argument merges
|
|
them onto the required action."""
|
|
|
|
def f(symbol: Literal["AAPL", "TSLA"]) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
extra_group = ArgparseArgumentGroupModel(
|
|
name="provider_b",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="symbol",
|
|
type=str,
|
|
dest="symbol",
|
|
default=None,
|
|
required=True,
|
|
action="store",
|
|
help=None,
|
|
nargs=None,
|
|
choices=("MSFT", "AMZN"),
|
|
)
|
|
],
|
|
)
|
|
t = ArgparseTranslator(f, custom_argument_groups=[extra_group])
|
|
parser = t.parser
|
|
for action in parser._actions:
|
|
if action.dest == "symbol":
|
|
choices = list(action.choices or [])
|
|
assert {"AAPL", "TSLA", "MSFT", "AMZN"}.issubset(set(choices))
|
|
return
|
|
pytest.fail("symbol action missing")
|
|
|
|
|
|
def test_handle_argument_in_groups_optional_existing_argument_merges_provider_help():
|
|
"""A second provider re-defining an *optional* argument extends the help text
|
|
with provider info."""
|
|
|
|
def f(provider: Literal["fmp"] = "fmp", limit: int = 100) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
extra_group = ArgparseArgumentGroupModel(
|
|
name="yfinance",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="limit",
|
|
type=int,
|
|
dest="limit",
|
|
default=50,
|
|
required=False,
|
|
action="store",
|
|
help="Limit override.",
|
|
nargs=None,
|
|
choices=None,
|
|
)
|
|
],
|
|
)
|
|
t = ArgparseTranslator(f, custom_argument_groups=[extra_group])
|
|
parser = t.parser
|
|
for action in parser._actions:
|
|
if action.dest == "limit":
|
|
assert action.help is not None
|
|
return
|
|
pytest.fail("limit action missing")
|
|
|
|
|
|
def test_handle_argument_in_groups_third_group_demotes_required_arg_to_optional():
|
|
"""When the same argument exists in multiple non-(required/optional) groups,
|
|
it gets removed and re-added to the optional arguments group."""
|
|
|
|
def f(symbol: Literal["AAPL"] = "AAPL") -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
group_a = ArgparseArgumentGroupModel(
|
|
name="provider_a",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="symbol",
|
|
type=str,
|
|
dest="symbol",
|
|
default="AAPL",
|
|
required=False,
|
|
action="store",
|
|
help="from a",
|
|
nargs=None,
|
|
choices=("X1",),
|
|
)
|
|
],
|
|
)
|
|
group_b = ArgparseArgumentGroupModel(
|
|
name="provider_b",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="symbol",
|
|
type=str,
|
|
dest="symbol",
|
|
default="AAPL",
|
|
required=False,
|
|
action="store",
|
|
help="from b",
|
|
nargs=None,
|
|
choices=("X2",),
|
|
)
|
|
],
|
|
)
|
|
t = ArgparseTranslator(f, custom_argument_groups=[group_a, group_b])
|
|
parser = t.parser
|
|
for action in parser._actions:
|
|
if action.dest == "symbol":
|
|
choices = set(action.choices or [])
|
|
assert {"AAPL", "X1", "X2"}.issubset(choices)
|
|
return
|
|
|
|
|
|
def test_get_base_type_picks_first_concrete_when_no_str_or_bool():
|
|
"""``Union[date, int]`` picks ``date`` (first non-None concrete type) — line 264-267."""
|
|
from datetime import date
|
|
|
|
def f(when: date | int = date.today()) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
base, _ = t._get_type_and_choices(param)
|
|
assert base is date
|
|
|
|
|
|
def test_get_base_type_for_any_returns_str():
|
|
"""``Any`` falls through to str."""
|
|
from typing import Any
|
|
|
|
def f(payload: Any = "x") -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
base, _ = t._get_type_and_choices(param)
|
|
assert base is str
|
|
|
|
|
|
def test_get_base_type_unknown_origin_returns_str():
|
|
"""A non-class, non-typing-origin annotation falls through to str."""
|
|
|
|
def f(value=None) -> dict:
|
|
"""f."""
|
|
return {}
|
|
|
|
t = ArgparseTranslator(f)
|
|
param = next(iter(t.signature.parameters.values()))
|
|
t.type_hints["value"] = object()
|
|
base, _ = t._get_type_and_choices(param)
|
|
assert base is str
|
|
|
|
|
|
def test_execute_func_with_provider_passes_provider_args():
|
|
"""``execute_func`` populates ``provider_args`` from the matching provider
|
|
bucket."""
|
|
|
|
def f(
|
|
symbol: str = "AAPL",
|
|
provider: Literal["fmp", "yf"] = "fmp",
|
|
interval: str | None = None,
|
|
) -> dict:
|
|
"""f."""
|
|
return {"symbol": symbol, "provider": provider, "interval": interval}
|
|
|
|
extra = ArgparseArgumentGroupModel(
|
|
name="fmp",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="interval",
|
|
type=str,
|
|
dest="interval",
|
|
default="1d",
|
|
required=False,
|
|
action="store",
|
|
help=None,
|
|
nargs=None,
|
|
choices=("1d", "1h"),
|
|
)
|
|
],
|
|
)
|
|
t = ArgparseTranslator(f, custom_argument_groups=[extra])
|
|
parsed = t.parser.parse_args(
|
|
["--symbol", "X", "--provider", "fmp", "--interval", "1h"]
|
|
)
|
|
out = t.execute_func(parsed)
|
|
assert out["symbol"] == "X"
|
|
assert out["interval"] == "1h"
|
|
|
|
|
|
def test_execute_func_without_provider_iterates_all_providers():
|
|
"""No matching provider → ``provider_args`` collects every group's params."""
|
|
|
|
def f(symbol: str = "AAPL", interval: str | None = None) -> dict:
|
|
"""f."""
|
|
return {"symbol": symbol, "interval": interval}
|
|
|
|
extra = ArgparseArgumentGroupModel(
|
|
name="fmp",
|
|
arguments=[
|
|
ArgparseArgumentModel(
|
|
name="interval",
|
|
type=str,
|
|
dest="interval",
|
|
default="1d",
|
|
required=False,
|
|
action="store",
|
|
help=None,
|
|
nargs=None,
|
|
choices=("1d", "1h"),
|
|
)
|
|
],
|
|
)
|
|
t = ArgparseTranslator(f, custom_argument_groups=[extra])
|
|
parsed = t.parser.parse_args(["--symbol", "X", "--interval", "1h"])
|
|
out = t.execute_func(parsed)
|
|
assert out["interval"] == "1h"
|
|
|
|
|
|
def test_parse_args_and_execute_invokes_func(monkeypatch):
|
|
"""``parse_args_and_execute`` parses ``sys.argv`` and runs ``execute_func``."""
|
|
import sys
|
|
|
|
def f(symbol: str = "AAPL") -> dict:
|
|
"""f."""
|
|
return {"symbol": symbol}
|
|
|
|
t = ArgparseTranslator(f)
|
|
monkeypatch.setattr(sys, "argv", ["prog", "--symbol", "MSFT"])
|
|
out = t.parse_args_and_execute()
|
|
assert out == {"symbol": "MSFT"}
|
|
|
|
|
|
def test_translate_returns_callable_that_runs_func(monkeypatch):
|
|
"""``translate()`` returns a no-arg wrapper that drives ``parse_args_and_execute``."""
|
|
import sys
|
|
|
|
def f(symbol: str = "AAPL") -> dict:
|
|
"""f."""
|
|
return {"symbol": symbol}
|
|
|
|
t = ArgparseTranslator(f)
|
|
wrapper = t.translate()
|
|
monkeypatch.setattr(sys, "argv", ["prog", "--symbol", "TSLA"])
|
|
out = wrapper()
|
|
assert out == {"symbol": "TSLA"}
|
|
|
|
|
|
def test_get_argument_custom_choices_via_openbbfield():
|
|
"""An ``OpenBBField`` annotation with ``choices=...`` overrides Literal choices."""
|
|
from typing import Annotated
|
|
|
|
from openbb_core.app.model.field import OpenBBField
|
|
|
|
def f(
|
|
symbol: Annotated[
|
|
str, OpenBBField(description="x", choices=["AAA", "BBB"])
|
|
] = "AAA",
|
|
) -> dict:
|
|
"""f."""
|
|
return {"symbol": symbol}
|
|
|
|
t = ArgparseTranslator(f)
|
|
parser = t.parser
|
|
for action in parser._actions:
|
|
if action.dest == "symbol":
|
|
assert set(action.choices) == {"AAA", "BBB"}
|
|
return
|
|
pytest.fail("symbol action missing")
|
|
|
|
|
|
def test_update_with_custom_types_skips_kwargs_param():
|
|
"""``_update_with_custom_types`` skips the ``**kwargs`` placeholder."""
|
|
|
|
def f(name: str = "x", **kwargs) -> dict:
|
|
"""f."""
|
|
return {"name": name, **kwargs}
|
|
|
|
t = ArgparseTranslator(f)
|
|
parsed = t.parser.parse_args(["--name", "y"])
|
|
out = t.execute_func(parsed)
|
|
assert out["name"] == "y"
|
|
|
|
|
|
def test_update_with_custom_types_constructs_basemodel_for_basemodel_param():
|
|
"""A BaseModel param has its flattened kwargs assembled into the model."""
|
|
t = ArgparseTranslator(custom_type_function)
|
|
parsed = t.parser.parse_args(["--data__field1", "hello", "--data__field2", "20"])
|
|
out = t.execute_func(parsed)
|
|
inner = out["data"]
|
|
assert isinstance(inner, CustomData)
|
|
assert inner.field1 == "hello"
|
|
assert inner.field2 == 20
|