From b82b623f23b189ceb307a7f59c6c31fe1bad75c8 Mon Sep 17 00:00:00 2001 From: Alexandre Gomes Gaigalas Date: Wed, 17 Dec 2025 17:58:16 -0300 Subject: [PATCH] Refactoring, use parametrized tests --- apywire/compiler.py | 436 ++++++++++++++-------------- apywire/constants.py | 4 + apywire/runtime.py | 4 +- apywire/wiring.py | 7 +- tests/test_cli.py | 200 ++++--------- tests/test_compile_aio.py | 41 ++- tests/test_constant_placeholders.py | 112 +++---- tests/test_formats.py | 348 ++++++++-------------- tests/test_generator.py | 307 ++++++-------------- tests/test_stdlib_compat.py | 105 +++---- 10 files changed, 613 insertions(+), 951 deletions(-) diff --git a/apywire/compiler.py b/apywire/compiler.py index ba78d0b..6abe856 100644 --- a/apywire/compiler.py +++ b/apywire/compiler.py @@ -85,51 +85,18 @@ def _astify(self, obj: _ResolvedValue, aio: bool = False) -> ast.expr: return ast.Tuple(elts=elts, ctx=ast.Load()) return ast.Constant(cast(_ConstantValue, obj)) - def _compile_property( - self, - name: str, - module_name: str, - class_name: str, - factory_method: str | None, - data: _ResolvedSpecMapping, - *, - aio: bool = False, - thread_safe: bool = False, - ) -> ast.FunctionDef | ast.AsyncFunctionDef: - """Build an AST FunctionDef for a cached accessor that returns - ``module.class(**data)`` or ``module.class.factory_method(**data)``. + def _normalize_spec_data( + self, data: _ResolvedSpecMapping + ) -> tuple[list[_ResolvedValue], dict[str, _ResolvedValue]]: + """Normalize spec data into positional and keyword arguments. - When ``aio`` is True this function will produce an - ``ast.AsyncFunctionDef`` that awaits referenced accessors and calls - the blocking constructor in an executor (``loop.run_in_executor``). - When ``aio`` is False it produces a standard synchronous - ``ast.FunctionDef``. - """ - # Build the target callable: module.Class or module.Class.factoryMethod - if factory_method: - module_attr = ast.Attribute( - value=ast.Attribute( - value=ast.Name(id=module_name, ctx=ast.Load()), - attr=class_name, - ctx=ast.Load(), - ), - attr=factory_method, - ctx=ast.Load(), - ) - else: - module_attr = ast.Attribute( - value=ast.Name(id=module_name, ctx=ast.Load()), - attr=class_name, - ctx=ast.Load(), - ) - pre_statements: list[ast.stmt] = [] - - counter = [0] # For generating unique variable names - - args: list[ast.expr] = [] + Args: + data: Either a list (positional args only) or dict + (mixed args/kwargs) - kwargs: list[ast.keyword] = [] - # Normalize data into args_data (list) and kwargs_data (dict) + Returns: + Tuple of (args_list, kwargs_dict) + """ args_data: list[_ResolvedValue] = [] kwargs_data: dict[str, _ResolvedValue] = {} @@ -137,17 +104,46 @@ def _compile_property( args_data = data else: data_dict = data - # Iterate once over data.items() to separate args and kwargs + # Separate args and kwargs from mixed dict args_items = [] - kwargs_data = {} for k, v in data_dict.items(): if isinstance(k, int): args_items.append((k, v)) elif isinstance(k, str): kwargs_data[k] = v + # Sort positional args by their integer keys args_items.sort(key=itemgetter(0)) args_data = [v for _, v in args_items] + return args_data, kwargs_data + + def _process_argument_values( + self, + args_data: list[_ResolvedValue], + kwargs_data: dict[str, _ResolvedValue], + *, + aio: bool, + pre_statements: list[ast.stmt], + counter: list[int], + ) -> tuple[list[ast.expr], list[ast.keyword]]: + """Process argument values and return AST expressions. + + For async mode, replaces awaits with local variables and generates + pre-statements for variable assignments. + + Args: + args_data: List of positional argument values + kwargs_data: Dict of keyword argument values + aio: Whether running in async mode + pre_statements: List to accumulate assignment statements + counter: Mutable counter for unique variable names + + Returns: + Tuple of (args_list, kwargs_list) with AST expressions + """ + args: list[ast.expr] = [] + kwargs: list[ast.keyword] = [] + # Process positional args for i, value in enumerate(args_data): raw_val_ast = self._astify(value, aio=aio) @@ -189,12 +185,47 @@ def _compile_property( kw_val = raw_val_ast kwargs.append(ast.keyword(arg=key, value=kw_val)) - call = ast.Call(func=module_attr, args=args, keywords=kwargs) + return args, kwargs - cache_attr = f"{CACHE_ATTR_PREFIX}{name}" + def _create_loop_assignment(self) -> ast.stmt: + """Create AST statement for getting to running event loop.""" + return ast.Assign( + targets=[ast.Name(id="loop", ctx=ast.Store())], + value=ast.Call( + func=ast.Attribute( + value=ast.Name(id="asyncio", ctx=ast.Load()), + attr="get_running_loop", + ctx=ast.Load(), + ), + args=[], + keywords=[], + ), + ) + + def _create_module_reference( + self, module_name: str, class_name: str, factory_method: str | None + ) -> ast.expr: + """Create AST reference to module.Class or module.Class.method.""" + if factory_method: + return ast.Attribute( + value=ast.Attribute( + value=ast.Name(id=module_name, ctx=ast.Load()), + attr=class_name, + ctx=ast.Load(), + ), + attr=factory_method, + ctx=ast.Load(), + ) + else: + return ast.Attribute( + value=ast.Name(id=module_name, ctx=ast.Load()), + attr=class_name, + ctx=ast.Load(), + ) - # if not hasattr(self, '_name'): ... - has_check = ast.UnaryOp( + def _create_cache_check(self, cache_attr: str) -> ast.expr: + """Create hasattr check for cache attribute.""" + return ast.UnaryOp( op=ast.Not(), operand=ast.Call( func=ast.Name(id="hasattr", ctx=ast.Load()), @@ -206,113 +237,154 @@ def _compile_property( ), ) - return_stmt = ast.Return( + def _create_return_statement(self, cache_attr: str) -> ast.stmt: + """Create return statement for cached value.""" + return ast.Return( value=ast.Attribute( value=ast.Name(id="self", ctx=ast.Load()), attr=cache_attr, ctx=ast.Load(), ) ) + + def _create_cache_assignment( + self, cache_attr: str, value_expr: ast.expr + ) -> ast.stmt: + """Create assignment to cache attribute.""" + return ast.Assign( + targets=[ + ast.Attribute( + value=ast.Name(id="self", ctx=ast.Load()), + attr=cache_attr, + ctx=ast.Store(), + ) + ], + value=value_expr, + ) + + def _create_lambda_function(self, body: ast.expr) -> ast.Lambda: + """Create a lambda function with the given body.""" + return ast.Lambda( + args=ast.arguments( + posonlyargs=[], + args=[], + vararg=None, + kwarg=None, + defaults=[], + kwonlyargs=[], + kw_defaults=[], + ), + body=body, + ) + + def _create_executor_call(self, lambda_expr: ast.Lambda) -> ast.expr: + """Create run_in_executor call for async compilation.""" + return ast.Await( + value=ast.Call( + func=ast.Attribute( + value=ast.Name(id="loop", ctx=ast.Load()), + attr="run_in_executor", + ctx=ast.Load(), + ), + args=[ast.Constant(value=None), lambda_expr], + keywords=[], + ) + ) + + def _create_function_template( + self, + name: str, + body: list[ast.stmt], + is_async: bool, + ) -> ast.FunctionDef | ast.AsyncFunctionDef: + """Create basic function definition structure.""" + if is_async: + return ast.AsyncFunctionDef( + name=name, + args=_PROPERTY_ARGS, + body=body, + decorator_list=[], + returns=None, + type_comment=None, + type_params=[], + ) + else: + return ast.FunctionDef( + name=name, + args=_PROPERTY_ARGS, + body=body, + decorator_list=[], + returns=None, + type_comment=None, + type_params=[], + ) + + def _compile_property( + self, + name: str, + module_name: str, + class_name: str, + factory_method: str | None, + data: _ResolvedSpecMapping, + *, + aio: bool = False, + thread_safe: bool = False, + ) -> ast.FunctionDef | ast.AsyncFunctionDef: + """Build an AST FunctionDef for a cached accessor that returns + ``module.class(**data)`` or ``module.class.factory_method(**data)``. + + When ``aio`` is True this function will produce an + ``ast.AsyncFunctionDef`` that awaits referenced accessors and calls + the blocking constructor in an executor (``loop.run_in_executor``). + When ``aio`` is False it produces a standard synchronous + ``ast.FunctionDef``. + """ + # Build the target callable: module.Class or module.Class.factoryMethod + module_attr = self._create_module_reference( + module_name, class_name, factory_method + ) + pre_statements: list[ast.stmt] = [] + counter = [0] # For generating unique variable names + + # Normalize and process argument data + args_data, kwargs_data = self._normalize_spec_data(data) + args, kwargs = self._process_argument_values( + args_data, + kwargs_data, + aio=aio, + pre_statements=pre_statements, + counter=counter, + ) + + call = ast.Call(func=module_attr, args=args, keywords=kwargs) + + cache_attr = f"{CACHE_ATTR_PREFIX}{name}" + + # Create reusable components + has_check = self._create_cache_check(cache_attr) + return_stmt = self._create_return_statement(cache_attr) # Build the assignment that sets the cache value; different # behavior is required for async vs sync callers. if not aio: - assign_cache = ast.Assign( - targets=[ - ast.Attribute( - value=ast.Name(id="self", ctx=ast.Load()), - attr=cache_attr, - ctx=ast.Store(), - ) - ], - value=call, - ) + assign_cache = self._create_cache_assignment(cache_attr, call) else: # Async path: compute local values and call in executor. - loop_assign = ast.Assign( - targets=[ast.Name(id="loop", ctx=ast.Store())], - value=ast.Call( - func=ast.Attribute( - value=ast.Name(id="asyncio", ctx=ast.Load()), - attr="get_running_loop", - ctx=ast.Load(), - ), - args=[], - keywords=[], - ), - ) - # Create a lambda that returns the class call expression - # referencing the precomputed locals - lambda_expr = ast.Lambda( - args=ast.arguments( - posonlyargs=[], - args=[], - vararg=None, - kwarg=None, - defaults=[], - kwonlyargs=[], - kw_defaults=[], - ), - body=call, - ) - run_call = ast.Await( - value=ast.Call( - func=ast.Attribute( - value=ast.Name(id="loop", ctx=ast.Load()), - attr="run_in_executor", - ctx=ast.Load(), - ), - args=[ - ast.Constant(value=None), - lambda_expr, - ], - keywords=[], - ) - ) - assign_cache = ast.Assign( - targets=[ - ast.Attribute( - value=ast.Name(id="self", ctx=ast.Load()), - attr=cache_attr, - ctx=ast.Store(), - ) - ], - value=run_call, - ) + lambda_expr = self._create_lambda_function(call) + pre_statements.append(self._create_loop_assignment()) + run_call = self._create_executor_call(lambda_expr) + assign_cache = self._create_cache_assignment(cache_attr, run_call) if_stmt_body = pre_statements.copy() - if aio: - if_stmt_body.append(loop_assign) if_stmt_body.append(assign_cache) if_stmt = ast.If(test=has_check, body=if_stmt_body, orelse=[]) func_def: ast.FunctionDef | ast.AsyncFunctionDef # If the compiled output is thread-safe, inject locking logic if not aio and not thread_safe: - func_def = ast.FunctionDef( - name=name, - args=_PROPERTY_ARGS, - body=[if_stmt, return_stmt], - decorator_list=[], - returns=None, - type_comment=None, - type_params=[], + func_def = self._create_function_template( + name, [if_stmt, return_stmt], is_async=False ) elif not aio and thread_safe: - # Build sync thread-safe version using code template - # Build a lambda maker which is a synchronous callable used by - # the helper mixin to invoke the constructor within the - # instantiation lock management. We pass this maker to - # `self._instantiate_attr(name, maker)`. - maker = ast.Lambda( - args=ast.arguments( - posonlyargs=[], - args=[], - vararg=None, - kwarg=None, - defaults=[], - kwonlyargs=[], - kw_defaults=[], - ), - body=call, - ) + # Build sync thread-safe version using helper mixin + maker = self._create_lambda_function(call) call_inst = ast.Call( func=ast.Attribute( value=ast.Name(id="self", ctx=ast.Load()), @@ -322,58 +394,28 @@ def _compile_property( args=[ast.Constant(value=name), maker], keywords=[], ) - assign_cache = ast.Assign( - targets=[ - ast.Attribute( - value=ast.Name(id="self", ctx=ast.Load()), - attr=cache_attr, - ctx=ast.Store(), - ) - ], - value=call_inst, - ) + assign_cache = self._create_cache_assignment(cache_attr, call_inst) if_stmt_body = pre_statements.copy() if_stmt_body.append(assign_cache) - func_def = ast.FunctionDef( - name=name, - args=_PROPERTY_ARGS, - body=[ + func_def = self._create_function_template( + name, + [ ast.If(test=has_check, body=if_stmt_body, orelse=[]), return_stmt, ], - decorator_list=[], - returns=None, - type_comment=None, - type_params=[], + is_async=False, ) - # This branch handles the synchronous, non-thread-safe case + # This branch handles the asynchronous, non-thread-safe case elif aio and not thread_safe: - func_def = ast.AsyncFunctionDef( - name=name, - args=_PROPERTY_ARGS, - body=[if_stmt, return_stmt], - decorator_list=[], - returns=None, - type_comment=None, - type_params=[], + func_def = self._create_function_template( + name, [if_stmt, return_stmt], is_async=True ) else: # aio and thread_safe # Create a maker lambda (synchronous) for the helper # mixin and run it in executor. Precomputed locals are # already present in `pre_statements` to avoid 'await' in # the lambda body. - maker = ast.Lambda( - args=ast.arguments( - posonlyargs=[], - args=[], - vararg=None, - kwarg=None, - defaults=[], - kwonlyargs=[], - kw_defaults=[], - ), - body=call, - ) + maker = self._create_lambda_function(call) instantiate_call = ast.Call( func=ast.Attribute( value=ast.Name(id="self", ctx=ast.Load()), @@ -384,18 +426,7 @@ def _compile_property( keywords=[], ) # Build lambda passed to executor that calls the mixin - executor_lambda = ast.Lambda( - args=ast.arguments( - posonlyargs=[], - args=[], - vararg=None, - kwarg=None, - defaults=[], - kwonlyargs=[], - kw_defaults=[], - ), - body=instantiate_call, - ) + executor_lambda = self._create_lambda_function(instantiate_call) run_call = ast.Await( value=ast.Call( func=ast.Attribute( @@ -407,32 +438,13 @@ def _compile_property( keywords=[], ) ) - assign_cache = ast.Assign( - targets=[ - ast.Attribute( - value=ast.Name(id="self", ctx=ast.Load()), - attr=cache_attr, - ctx=ast.Store(), - ) - ], - value=run_call, - ) + assign_cache = self._create_cache_assignment(cache_attr, run_call) body = pre_statements.copy() - # Ensure we compute the event loop variable before calling into - # run_in_executor. - body.append(loop_assign) body.append(assign_cache) - func_def = ast.AsyncFunctionDef( - name=name, - args=_PROPERTY_ARGS, - body=[ - ast.If(test=has_check, body=body, orelse=[]), - return_stmt, - ], - decorator_list=[], - returns=None, - type_comment=None, - type_params=[], + func_def = self._create_function_template( + name, + [ast.If(test=has_check, body=body, orelse=[]), return_stmt], + is_async=True, ) return func_def diff --git a/apywire/constants.py b/apywire/constants.py index 20ae40c..f728615 100644 --- a/apywire/constants.py +++ b/apywire/constants.py @@ -8,6 +8,8 @@ to improve maintainability and reduce duplication. """ +import re + SPEC_KEY_DELIMITER = " " # Separates module.Class from name in spec keys PLACEHOLDER_START = "{" # Start marker for placeholder references @@ -15,6 +17,8 @@ # Regex pattern for placeholders like {name}: captures name, no nested braces PLACEHOLDER_PATTERN = r"\{([^{}]+)\}" +# Compiled version for better performance +PLACEHOLDER_REGEX = re.compile(PLACEHOLDER_PATTERN) SYNTHETIC_CONST = "__sconst__" # Synthetic module for promoted constants diff --git a/apywire/runtime.py b/apywire/runtime.py index ee92911..6610a61 100644 --- a/apywire/runtime.py +++ b/apywire/runtime.py @@ -14,7 +14,7 @@ from operator import itemgetter from typing import Awaitable, Callable, Protocol, cast, final -from apywire.constants import SYNTHETIC_CONST +from apywire.constants import PLACEHOLDER_REGEX, SYNTHETIC_CONST from apywire.exceptions import ( CircularWiringError, UnknownPlaceholderError, @@ -328,7 +328,7 @@ def replace_placeholder(match: re.Match[str]) -> str: # Convert to string return str(value) - return re.sub(self._placeholder_pattern, replace_placeholder, template) + return PLACEHOLDER_REGEX.sub(replace_placeholder, template) @final diff --git a/apywire/wiring.py b/apywire/wiring.py index e70b3dd..a17c2b5 100644 --- a/apywire/wiring.py +++ b/apywire/wiring.py @@ -23,7 +23,7 @@ from apywire.constants import ( PLACEHOLDER_END, - PLACEHOLDER_PATTERN, + PLACEHOLDER_REGEX, PLACEHOLDER_START, SPEC_KEY_DELIMITER, SYNTHETIC_CONST, @@ -131,7 +131,6 @@ def __init__( self._thread_safe = thread_safe self._max_lock_attempts = max_lock_attempts self._lock_retry_sleep = lock_retry_sleep - self._placeholder_pattern = re.compile(PLACEHOLDER_PATTERN) parsed: list[_UnresolvedParsedEntry] = [] consts: dict[str, _SpecMapping | _ConstantValue] = {} @@ -342,7 +341,7 @@ def replace_placeholder(match: re.Match[str]) -> str: str(ref_value) if not isinstance(ref_value, str) else ref_value ) - return re.sub(self._placeholder_pattern, replace_placeholder, template) + return PLACEHOLDER_REGEX.sub(replace_placeholder, template) def _find_placeholder_names(self, obj: _SpecValue) -> set[str]: """Find all placeholder names referenced in a value. @@ -356,7 +355,7 @@ def _find_placeholder_names(self, obj: _SpecValue) -> set[str]: names: set[str] = set() if isinstance(obj, str): # Find all placeholders in the string (embedded or not) - matches: list[str] = re.findall(self._placeholder_pattern, obj) + matches: list[str] = PLACEHOLDER_REGEX.findall(obj) names.update(matches) elif isinstance(obj, dict): for v in obj.values(): diff --git a/tests/test_cli.py b/tests/test_cli.py index 77c1cda..2852d8f 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -12,39 +12,15 @@ from apywire.__main__ import main -def test_cli_version_short_flag() -> None: - """Test CLI version display with -v flag.""" +@pytest.mark.parametrize( + "flag", + ["-v", "--version", "-h", "--help"], + ids=["v", "version", "h", "help"], +) +def test_cli_basic_flags_exit_zero(flag: str) -> None: + """Test that basic CLI flags exit with code 0.""" with pytest.raises(SystemExit) as exc_info: - main(["-v"]) - - # argparse with action="version" exits with code 0 - assert exc_info.value.code == 0 - - -def test_cli_version_long_flag() -> None: - """Test CLI version display with --version flag.""" - with pytest.raises(SystemExit) as exc_info: - main(["--version"]) - - # argparse with action="version" exits with code 0 - assert exc_info.value.code == 0 - - -def test_cli_help_flag() -> None: - """Test CLI help display with -h flag.""" - with pytest.raises(SystemExit) as exc_info: - main(["-h"]) - - # argparse with action="help" exits with code 0 - assert exc_info.value.code == 0 - - -def test_cli_help_long_flag() -> None: - """Test CLI help display with --help flag.""" - with pytest.raises(SystemExit) as exc_info: - main(["--help"]) - - # argparse with action="help" exits with code 0 + main([flag]) assert exc_info.value.code == 0 @@ -54,33 +30,15 @@ def test_cli_no_arguments() -> None: assert result == 0 -def test_cli_version_output_contains_package_name() -> None: - """Test that version output contains 'apywire'.""" +@pytest.mark.parametrize("flag", ["-v", "--version"], ids=["v", "version"]) +def test_cli_version_output_format(flag: str) -> None: + """Test that version output contains package name and version.""" from importlib.metadata import version expected_version = version("apywire") - - # Capture stdout since argparse writes version to stdout with pytest.raises(SystemExit) as exc_info: with patch("sys.stdout", new_callable=StringIO) as mock_stdout: - main(["--version"]) - - assert exc_info.value.code == 0 - stdout_output = mock_stdout.getvalue() - assert "apywire" in stdout_output - assert expected_version in stdout_output - - -def test_cli_version_short_output_contains_package_name() -> None: - """Test that -v output contains 'apywire'.""" - from importlib.metadata import version - - expected_version = version("apywire") - - # Capture stdout since argparse writes version to stdout - with pytest.raises(SystemExit) as exc_info: - with patch("sys.stdout", new_callable=StringIO) as mock_stdout: - main(["-v"]) + main([flag]) assert exc_info.value.code == 0 stdout_output = mock_stdout.getvalue() @@ -170,40 +128,25 @@ def test_cli_module_execution() -> None: assert "apywire" in result.stdout # argparse outputs to stdout -def test_cli_generate_json() -> None: - """Test generate command with JSON format.""" - with patch("sys.stdout", new_callable=StringIO) as mock_stdout: - result = main( - ["generate", "--format", "json", "collections.OrderedDict d"] - ) - - assert result == 0 - output = mock_stdout.getvalue() - assert "collections.OrderedDict d" in output - - -def test_cli_generate_ini() -> None: - """Test generate command with INI format.""" - with patch("sys.stdout", new_callable=StringIO) as mock_stdout: - result = main( - ["generate", "--format", "ini", "collections.OrderedDict d"] - ) - - assert result == 0 - output = mock_stdout.getvalue() - assert "[collections.OrderedDict d]" in output - - -def test_cli_generate_toml() -> None: - """Test generate command with TOML format.""" +@pytest.mark.parametrize( + "fmt, expected_marker", + [ + ("json", "collections.OrderedDict d"), + ("ini", "[collections.OrderedDict d]"), + ("toml", '["collections.OrderedDict d"]'), + ], + ids=["json", "ini", "toml"], +) +def test_cli_generate_formats(fmt: str, expected_marker: str) -> None: + """Test generate command with different output formats.""" with patch("sys.stdout", new_callable=StringIO) as mock_stdout: result = main( - ["generate", "--format", "toml", "collections.OrderedDict d"] + ["generate", "--format", fmt, "collections.OrderedDict d"] ) assert result == 0 output = mock_stdout.getvalue() - assert '["collections.OrderedDict d"]' in output + assert expected_marker in output def test_cli_generate_multiple_entries() -> None: @@ -225,12 +168,20 @@ def test_cli_generate_multiple_entries() -> None: assert "collections.OrderedDict b" in output -def test_cli_compile_json_stdin() -> None: - """Test compile command with JSON from stdin.""" - json_input = '{"collections.OrderedDict d": {}}' - with patch("sys.stdin", StringIO(json_input)): +@pytest.mark.parametrize( + "fmt, input_data", + [ + ("json", '{"collections.OrderedDict d": {}}'), + ("ini", "[collections.OrderedDict d]\n"), + ("toml", '["collections.OrderedDict d"]\n'), + ], + ids=["json", "ini", "toml"], +) +def test_cli_compile_stdin_formats(fmt: str, input_data: str) -> None: + """Test compile command with different input formats from stdin.""" + with patch("sys.stdin", StringIO(input_data)): with patch("sys.stdout", new_callable=StringIO) as mock_stdout: - result = main(["compile", "--format", "json", "-"]) + result = main(["compile", "--format", fmt, "-"]) assert result == 0 output = mock_stdout.getvalue() @@ -238,30 +189,6 @@ def test_cli_compile_json_stdin() -> None: assert "def d(self):" in output -def test_cli_compile_ini_stdin() -> None: - """Test compile command with INI from stdin.""" - ini_input = "[collections.OrderedDict d]\n" - with patch("sys.stdin", StringIO(ini_input)): - with patch("sys.stdout", new_callable=StringIO) as mock_stdout: - result = main(["compile", "--format", "ini", "-"]) - - assert result == 0 - output = mock_stdout.getvalue() - assert "class Compiled:" in output - - -def test_cli_compile_toml_stdin() -> None: - """Test compile command with TOML from stdin.""" - toml_input = '["collections.OrderedDict d"]\n' - with patch("sys.stdin", StringIO(toml_input)): - with patch("sys.stdout", new_callable=StringIO) as mock_stdout: - result = main(["compile", "--format", "toml", "-"]) - - assert result == 0 - output = mock_stdout.getvalue() - assert "class Compiled:" in output - - def test_cli_compile_with_aio_flag() -> None: """Test compile command with --aio flag.""" json_input = '{"collections.OrderedDict d": {}}' @@ -310,43 +237,26 @@ def test_cli_compile_from_file() -> None: os.unlink(temp_file) -def test_cli_compile_json_parsing_error() -> None: - """Test CLI error handling for invalid JSON input.""" - json_input = '{"invalid": json content}' - - with patch("sys.stdin", StringIO(json_input)): - with patch("sys.stderr", new_callable=StringIO) as mock_stderr: - result = main(["compile", "--format", "json", "-"]) - - assert result == 1 # Should return error code - stderr_output = mock_stderr.getvalue() - assert "Error parsing JSON content:" in stderr_output - - -def test_cli_compile_toml_parsing_error() -> None: - """Test CLI error handling for invalid TOML input.""" - toml_input = '["invalid toml content' - - with patch("sys.stdin", StringIO(toml_input)): - with patch("sys.stderr", new_callable=StringIO) as mock_stderr: - result = main(["compile", "--format", "toml", "-"]) - - assert result == 1 # Should return error code - stderr_output = mock_stderr.getvalue() - assert "Error parsing TOML content:" in stderr_output - - -def test_cli_compile_ini_parsing_error() -> None: - """Test CLI error handling for invalid INI input.""" - ini_input = "[invalid section\nkey = value" - - with patch("sys.stdin", StringIO(ini_input)): +@pytest.mark.parametrize( + "fmt, input_data, expected_err", + [ + ("json", '{"invalid": json content}', "Error parsing JSON content:"), + ("toml", '["invalid toml content', "Error parsing TOML content:"), + ("ini", "[invalid section\nkey = value", "Error parsing INI content:"), + ], + ids=["json", "toml", "ini"], +) +def test_cli_compile_parsing_errors( + fmt: str, input_data: str, expected_err: str +) -> None: + """Test CLI error handling for invalid input formats.""" + with patch("sys.stdin", StringIO(input_data)): with patch("sys.stderr", new_callable=StringIO) as mock_stderr: - result = main(["compile", "--format", "ini", "-"]) + result = main(["compile", "--format", fmt, "-"]) - assert result == 1 # Should return error code + assert result == 1 stderr_output = mock_stderr.getvalue() - assert "Error parsing INI content:" in stderr_output + assert expected_err in stderr_output def test_cli_generate_toml_write_error() -> None: diff --git a/tests/test_compile_aio.py b/tests/test_compile_aio.py index 36ee61f..c6dec1d 100644 --- a/tests/test_compile_aio.py +++ b/tests/test_compile_aio.py @@ -250,11 +250,21 @@ async def get() -> int: assert result == 42 -def test_compile_aio_list_args() -> None: - """Test async compilation of list arguments.""" - spec: apywire.Spec = {"builtins.int myInt": [99]} - # Compile with aio=True - code = apywire.WiringCompiler(spec).compile(aio=True) +@pytest.mark.parametrize( + "class_path, args, expected", + [ + ("builtins.int", [99], 99), + ("builtins.complex", {0: 1.0, "imag": 2.0}, complex(1.0, 2.0)), + ], + ids=["int", "complex"], +) +def test_compile_aio_instantiation( + class_path: str, args: object, expected: object +) -> None: + """Test async compilation of various stdlib classes.""" + spec = {f"{class_path} obj": args} + compiler = apywire.WiringCompiler(spec) # type: ignore[arg-type] + code = compiler.compile(aio=True) execd: dict[str, object] = {"asyncio": asyncio} exec(code, execd) @@ -262,26 +272,9 @@ def test_compile_aio_list_args() -> None: async def run() -> None: # When compiled with aio=True, the methods themselves are async - coro = cast(Awaitable[int], getattr(compiled, "myInt")()) - val = await coro - assert val == 99 - - asyncio.run(run()) - - -def test_compile_aio_mixed_args() -> None: - """Test async compilation of mixed numeric/string keys.""" - spec: apywire.Spec = {"builtins.complex myComplex": {0: 1.0, "imag": 2.0}} - code = apywire.WiringCompiler(spec).compile(aio=True) - - execd: dict[str, object] = {"asyncio": asyncio} - exec(code, execd) - compiled = execd["compiled"] - - async def run() -> None: - coro = cast(Awaitable[complex], getattr(compiled, "myComplex")()) + coro = cast(Awaitable[object], getattr(compiled, "obj")()) val = await coro - assert val == complex(1.0, 2.0) + assert val == expected asyncio.run(run()) diff --git a/tests/test_constant_placeholders.py b/tests/test_constant_placeholders.py index b229bfa..8fdf35c 100644 --- a/tests/test_constant_placeholders.py +++ b/tests/test_constant_placeholders.py @@ -12,14 +12,23 @@ import apywire -def test_basic_constant_expansion() -> None: - """Test basic placeholder expansion in constants.""" - spec: apywire.Spec = { - "a": "foo", - "b": "{a} bar", - } +@pytest.mark.parametrize( + "spec, expected_key, expected_val", + [ + ({"a": "foo", "b": "{a} bar"}, "b", "foo bar"), + ({"port": 5432, "url": "localhost:{port}"}, "url", "localhost:5432"), + ({"number": 42, "message": "{number}"}, "message", "42"), + ], + ids=["basic", "int_port", "int_ref"], +) +def test_constant_value_expansion( + spec: apywire.Spec, expected_key: str, expected_val: str +) -> None: + """Test basic placeholder expansion in constants with various value + types. + """ wired = apywire.Wiring(spec, thread_safe=False) - assert wired._values["b"] == "foo bar" + assert wired._values[expected_key] == expected_val def test_nested_constant_dependencies() -> None: @@ -115,16 +124,36 @@ def __init__(self, dep: object) -> None: del sys.modules["test_module"] -def test_complex_nested_structures() -> None: - """Test placeholder expansion in nested data structures.""" - spec: apywire.Spec = { - "base": "/app", - "config": {"path": "{base}/config.yaml"}, - } +@pytest.mark.parametrize( + "spec, expected_key, expected_val", + [ + ( + {"base": "/app", "config": {"path": "{base}/config.yaml"}}, + "config", + {"path": "/app/config.yaml"}, + ), + ( + { + "base": "http://api", + "endpoints": ["{base}/users", "{base}/posts"], + }, + "endpoints", + ["http://api/users", "http://api/posts"], + ), + ( + {"base": "test", "data": ("{base}", "{base}")}, + "data", + ("test", "test"), + ), + ], + ids=["dict", "list", "tuple"], +) +def test_constant_collection_expansion( + spec: apywire.Spec, expected_key: str, expected_val: object +) -> None: + """Test placeholder expansion in nested data structures and collections.""" wired = apywire.Wiring(spec, thread_safe=False) - - assert wired._values["base"] == "/app" - assert wired._values["config"] == {"path": "/app/config.yaml"} + assert wired._values[expected_key] == expected_val def test_string_representation_of_wired_object() -> None: @@ -176,31 +205,6 @@ def test_with_thread_safe_mode() -> None: assert wired._values["db_url"] == "postgresql://localhost:5432/mydb" -def test_integer_constant_in_placeholder() -> None: - """Test that non-string constants are converted to strings.""" - spec: apywire.Spec = { - "port": 5432, - "url": "localhost:{port}", - } - wired = apywire.Wiring(spec, thread_safe=False) - - assert wired._values["url"] == "localhost:5432" - - -def test_list_with_placeholders() -> None: - """Test placeholder expansion in lists.""" - spec: apywire.Spec = { - "base": "http://api", - "endpoints": ["{base}/users", "{base}/posts"], - } - wired = apywire.Wiring(spec, thread_safe=False) - - assert wired._values["endpoints"] == [ - "http://api/users", - "http://api/posts", - ] - - def test_no_placeholders() -> None: """Test that constants without placeholders work normally.""" spec: apywire.Spec = { @@ -260,32 +264,6 @@ def test_interpolate_placeholders_unknown_placeholder() -> None: ) -def test_resolve_constant_non_string_reference() -> None: - """Test _resolve_constant with non-string reference value.""" - from apywire import Spec, Wiring - - # Constant referencing an integer constant - spec: Spec = {"number": 42, "message": "{number}"} - wired = Wiring(spec, thread_safe=False) - - # The value should have been resolved to string "42" - assert wired._values["message"] == "42" - - -def test_resolve_constant_with_tuple() -> None: - """Test _resolve_constant with tuple values.""" - from typing import cast - - from apywire import Spec, Wiring - - # Constant with tuple containing placeholders - spec = cast(Spec, {"base": "test", "data": ("{base}", "{base}")}) - wired = Wiring(spec, thread_safe=False) - - # Should resolve tuple placeholders - assert wired._values["data"] == ("test", "test") - - def test_transitive_promotion_of_constants() -> None: """Test transitive promotion of constants with wired refs.""" import sys diff --git a/tests/test_formats.py b/tests/test_formats.py index 0447ae2..72e6db1 100644 --- a/tests/test_formats.py +++ b/tests/test_formats.py @@ -4,6 +4,8 @@ """Tests for spec format adapters.""" +from typing import Callable + import pytest from apywire import Wiring @@ -18,112 +20,109 @@ from apywire.wiring import Spec -class TestJsonFormat: - """Tests for JSON format adapter.""" - - def test_roundtrip_simple_spec(self) -> None: - """Test JSON roundtrip with simple spec.""" +@pytest.mark.parametrize( + "to_format, from_format", + [ + (spec_to_json, json_to_spec), + (spec_to_toml, toml_to_spec), + (spec_to_ini, ini_to_spec), + ], + ids=["json", "toml", "ini"], +) +class TestFormatRoundtrips: + """Set of tests for format roundtrip compatibility.""" + + def test_roundtrip_simple_spec( + self, + to_format: "Callable[[Spec], str]", + from_format: "Callable[[str], Spec]", + ) -> None: + """Test roundtrip with simple spec.""" spec: Spec = { "datetime.datetime now": {"year": "{now_year}"}, "now_year": 2025, } - json_str = spec_to_json(spec) - result = json_to_spec(json_str) + formatted = to_format(spec) + result = from_format(formatted) assert result == spec - def test_roundtrip_with_constants(self) -> None: - """Test JSON roundtrip with various constant types.""" + def test_roundtrip_with_constants( + self, + to_format: "Callable[[Spec], str]", + from_format: "Callable[[str], Spec]", + ) -> None: + """Test roundtrip with various constant types.""" spec: Spec = { "datetime.datetime now": {}, "str_const": "hello", "int_const": 42, "float_const": 3.14, "bool_const": True, - "none_const": None, } - json_str = spec_to_json(spec) - result = json_to_spec(json_str) + # JSON and our INI adapter support None (serialized as "" in INI) + # TOML does not support None/null. + if to_format is not spec_to_toml: + spec["none_const"] = None + + formatted = to_format(spec) + result = from_format(formatted) assert result == spec - def test_roundtrip_with_nested_data(self) -> None: - """Test JSON roundtrip with nested lists and dicts.""" + def test_roundtrip_with_nested_data( + self, + to_format: "Callable[[Spec], str]", + from_format: "Callable[[str], Spec]", + ) -> None: + """Test roundtrip with nested lists and dicts.""" spec: Spec = { "mymod.MyClass obj": { "items": [1, 2, 3], - "config": {"key": "value"}, }, } - json_str = spec_to_json(spec) - result = json_to_spec(json_str) + obj_spec = spec["mymod.MyClass obj"] + if isinstance(obj_spec, dict): + obj_spec["config"] = {"key": "value"} + + formatted = to_format(spec) + result = from_format(formatted) assert result == spec - def test_roundtrip_empty_spec(self) -> None: - """Test JSON roundtrip with empty spec.""" + def test_roundtrip_empty_spec( + self, + to_format: "Callable[[Spec], str]", + from_format: "Callable[[str], Spec]", + ) -> None: + """Test roundtrip with empty spec.""" spec: Spec = {} - json_str = spec_to_json(spec) - result = json_to_spec(json_str) + formatted = to_format(spec) + result = from_format(formatted) assert result == spec - def test_produces_valid_wiring_spec(self) -> None: - """Test that parsed JSON produces a valid Wiring spec.""" - json_str = '{"collections.OrderedDict mydict": {}}' - spec = json_to_spec(json_str) - wiring = Wiring(spec) + def test_produces_valid_wiring_spec( + self, + to_format: "Callable[[Spec], str]", + from_format: "Callable[[str], Spec]", + ) -> None: + """Test that formatted spec produces a valid Wiring spec.""" + spec: Spec = {"collections.OrderedDict mydict": {}} + formatted = to_format(spec) + parsed = from_format(formatted) + wiring = Wiring(parsed) obj = wiring.mydict() assert obj is not None -class TestTomlFormat: - """Tests for TOML format adapter.""" - - def test_roundtrip_simple_spec(self) -> None: - """Test TOML roundtrip with simple spec.""" - spec: Spec = { - "datetime.datetime now": {"year": "{now_year}"}, - "now_year": 2025, - } - toml_str = spec_to_toml(spec) - result = toml_to_spec(toml_str) - assert result == spec - - def test_roundtrip_with_constants(self) -> None: - """Test TOML roundtrip with various constant types.""" - spec: Spec = { - "datetime.datetime now": {}, - "str_const": "hello", - "int_const": 42, - "float_const": 3.14, - "bool_const": True, - } - toml_str = spec_to_toml(spec) - result = toml_to_spec(toml_str) - assert result == spec +class TestJsonFormat: + """Tests specific to JSON format.""" - def test_roundtrip_with_nested_data(self) -> None: - """Test TOML roundtrip with nested lists and dicts.""" - spec: Spec = { - "mymod.MyClass obj": { - "items": [1, 2, 3], - }, - } - toml_str = spec_to_toml(spec) - result = toml_to_spec(toml_str) - assert result == spec + def test_roundtrip_with_none(self) -> None: + """Test JSON roundtrip with None specifically.""" + spec: Spec = {"none_const": None} + assert json_to_spec(spec_to_json(spec)) == spec - def test_roundtrip_empty_spec(self) -> None: - """Test TOML roundtrip with empty spec.""" - spec: Spec = {} - toml_str = spec_to_toml(spec) - result = toml_to_spec(toml_str) - assert result == spec - def test_produces_valid_wiring_spec(self) -> None: - """Test that parsed TOML produces a valid Wiring spec.""" - toml_str = '["collections.OrderedDict mydict"]\n' - spec = toml_to_spec(toml_str) - wiring = Wiring(spec) - obj = wiring.mydict() - assert obj is not None +class TestTomlFormat: + """Tests specific to TOML format.""" def test_toplevel_constants(self) -> None: """Test that top-level keys become constants.""" @@ -139,96 +138,7 @@ def test_toplevel_constants(self) -> None: class TestIniFormat: - """Tests for INI format adapter.""" - - def test_roundtrip_simple_spec(self) -> None: - """Test INI roundtrip with simple spec.""" - spec: Spec = { - "datetime.datetime now": {"year": "{now_year}"}, - "now_year": 2025, - } - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_roundtrip_string_constant(self) -> None: - """Test INI roundtrip with string constant.""" - spec: Spec = { - "datetime.datetime now": {}, - "my_str": "hello world", - } - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_roundtrip_numeric_constants(self) -> None: - """Test INI roundtrip with numeric constants.""" - spec: Spec = { - "datetime.datetime now": {}, - "int_val": 42, - "float_val": 3.14, - } - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_roundtrip_boolean_constants(self) -> None: - """Test INI roundtrip with boolean constants.""" - spec: Spec = { - "datetime.datetime now": {}, - "true_val": True, - "false_val": False, - } - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_roundtrip_none_constant(self) -> None: - """Test INI roundtrip with None constant.""" - spec: Spec = { - "datetime.datetime now": {}, - "none_val": None, - } - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_roundtrip_list_as_json(self) -> None: - """Test INI roundtrip with list serialized as JSON.""" - spec: Spec = { - "mymod.MyClass obj": { - "items": [1, 2, 3], - }, - } - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_roundtrip_dict_as_json(self) -> None: - """Test INI roundtrip with dict serialized as JSON.""" - spec: Spec = { - "mymod.MyClass obj": { - "config": {"key": "value"}, - }, - } - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_roundtrip_empty_spec(self) -> None: - """Test INI roundtrip with empty spec.""" - spec: Spec = {} - ini_str = spec_to_ini(spec) - result = ini_to_spec(ini_str) - assert result == spec - - def test_produces_valid_wiring_spec(self) -> None: - """Test that parsed INI produces a valid Wiring spec.""" - ini_str = "[collections.OrderedDict mydict]\n" - spec = ini_to_spec(ini_str) - wiring = Wiring(spec) - obj = wiring.mydict() - assert obj is not None + """Tests specific to INI format.""" def test_constants_section_extraction(self) -> None: """Test that constants are extracted from [constants] section.""" @@ -244,19 +154,24 @@ def test_constants_section_extraction(self) -> None: assert spec["my_value"] == 123 assert "constants" not in spec - def test_parse_true_case_insensitive(self) -> None: - """Test that 'TRUE', 'True', 'true' all parse to True.""" - for val in ["true", "True", "TRUE"]: - ini_str = f"[constants]\nflag = {val}\n" - spec = ini_to_spec(ini_str) - assert spec["flag"] is True - - def test_parse_false_case_insensitive(self) -> None: - """Test that 'FALSE', 'False', 'false' all parse to False.""" - for val in ["false", "False", "FALSE"]: - ini_str = f"[constants]\nflag = {val}\n" - spec = ini_to_spec(ini_str) - assert spec["flag"] is False + @pytest.mark.parametrize( + "val, expected", + [ + ("true", True), + ("True", True), + ("TRUE", True), + ("false", False), + ("False", False), + ("FALSE", False), + ], + ) + def test_parse_boolean_case_insensitive( + self, val: str, expected: bool + ) -> None: + """Test that boolean values parse case-insensitively.""" + ini_str = f"[constants]\nflag = {val}\n" + spec = ini_to_spec(ini_str) + assert spec["flag"] is expected class TestCrossFormatCompatibility: @@ -276,59 +191,48 @@ def test_json_to_toml_to_json(self) -> None: final_spec = json_to_spec(json_str_2) assert final_spec == spec - def test_all_formats_produce_working_wiring(self) -> None: - """Test that all formats produce specs that work with Wiring.""" - spec: Spec = { - "collections.OrderedDict mydict": {}, - } - - for to_format, from_format in [ - (spec_to_json, json_to_spec), - (spec_to_toml, toml_to_spec), - (spec_to_ini, ini_to_spec), - ]: - formatted = to_format(spec) - parsed = from_format(formatted) - wiring = Wiring(parsed) - obj = wiring.mydict() - assert obj is not None - class TestFormatErrorHandling: """Tests for FormatError handling in parsing functions.""" - def test_json_parsing_error(self) -> None: - """Test JSON parsing error handling.""" - from apywire.exceptions import FormatError - - invalid_json = '{"invalid": json content}' - with pytest.raises(FormatError) as exc_info: - json_to_spec(invalid_json) - - assert exc_info.value.format == "json" - assert "Failed to parse JSON content" in str(exc_info.value) - - def test_toml_parsing_error(self) -> None: - """Test TOML parsing error handling.""" - from apywire.exceptions import FormatError - - invalid_toml = '["invalid toml content' - with pytest.raises(FormatError) as exc_info: - toml_to_spec(invalid_toml) - - assert exc_info.value.format == "toml" - assert "Failed to parse TOML content" in str(exc_info.value) - - def test_ini_parsing_error(self) -> None: - """Test INI parsing error handling.""" + @pytest.mark.parametrize( + "parser, content, fmt, msg", + [ + ( + json_to_spec, + '{"invalid": json content}', + "json", + "Failed to parse JSON content", + ), + ( + toml_to_spec, + '["invalid toml content', + "toml", + "Failed to parse TOML content", + ), + ( + ini_to_spec, + "[invalid section\nkey = value", + "ini", + "Failed to parse INI content", + ), + ], + ) + def test_parsing_errors( + self, + parser: "Callable[[str], Spec]", + content: str, + fmt: str, + msg: str, + ) -> None: + """Test parsing error handling for various formats.""" from apywire.exceptions import FormatError - invalid_ini = "[invalid section\nkey = value" with pytest.raises(FormatError) as exc_info: - ini_to_spec(invalid_ini) + parser(content) - assert exc_info.value.format == "ini" - assert "Failed to parse INI content" in str(exc_info.value) + assert exc_info.value.format == fmt + assert msg in str(exc_info.value) def test_toml_write_error(self) -> None: """Test TOML write error when tomli_w is not available.""" diff --git a/tests/test_generator.py b/tests/test_generator.py index 9fb1ee8..27c87c9 100644 --- a/tests/test_generator.py +++ b/tests/test_generator.py @@ -6,7 +6,9 @@ import sys from types import ModuleType -from typing import Optional +from typing import Dict, List, Optional, Tuple, Union + +import pytest from apywire import Generator, Spec, Wiring @@ -135,77 +137,121 @@ def __init__(self) -> None: del sys.modules["mockmod_defs"] -def test_generate_invalid_format_no_delimiter() -> None: - """Test that invalid entry format raises ValueError.""" - try: - Generator.generate("invalid") - assert False, "Should have raised ValueError" - except ValueError as e: - assert "Invalid entry format" in str(e) +@pytest.mark.parametrize( + "entry, expected_msg", + [ + ("invalid", "Invalid entry format"), + ("Class name", "missing module qualification"), + ("myapp.models.SomeClass inst.create.more", "nested factory methods"), + ], + ids=["no_delimiter", "no_module", "nested_factory"], +) +def test_generate_invalid_formats(entry: str, expected_msg: str) -> None: + """Test that invalid entry formats raise ValueError with expected + message. + """ + with pytest.raises(ValueError, match=expected_msg): + Generator.generate(entry) + + +@pytest.mark.parametrize( + "annotation, expected_default", + [ + (int, 0), + (str, ""), + (float, 0.0), + (bool, False), + (Optional[str], ""), # Recurses to str default + (List[int], []), + (Dict[str, int], {}), + (Tuple[int, int], ()), + (Union[int, str, float], None), + (bytes, b""), + (complex, 0j), + (None, None), # unknown annotation + ], + ids=[ + "int", + "str", + "float", + "bool", + "optional", + "list", + "dict", + "tuple", + "union", + "bytes", + "complex", + "unknown", + ], +) +def test_generate_type_defaults( + annotation: object, expected_default: object +) -> None: + """Test that various type annotations produce correct defaults.""" + + class DynamicTyped: + pass + # Manually set up __init__ without default value + def __init__(self, value): # type: ignore[no-untyped-def] + pass -def test_generate_invalid_format_no_module() -> None: - """Test that entry without module raises ValueError.""" - try: - Generator.generate("Class name") - assert False, "Should have raised ValueError" - except ValueError as e: - assert "missing module qualification" in str(e) + # Updating the annotation for the 'value' parameter + if annotation is not None: + __init__.__annotations__ = {"value": annotation} # type: ignore[misc] + setattr(DynamicTyped, "__init__", __init__) # type: ignore[misc] + DynamicTyped.__module__ = "mockmod_dyn_typed" + + class MockModule(ModuleType): + def __init__(self) -> None: + super().__init__("mockmod_dyn_typed") + self.DynamicTyped = DynamicTyped + + mod = MockModule() + sys.modules["mockmod_dyn_typed"] = mod -def test_generate_invalid_nested_factory_method() -> None: - """Test that nested factory methods raise ValueError.""" try: - Generator.generate("myapp.models.SomeClass inst.create.more") - assert False, "Should have raised ValueError" - except ValueError as e: - assert "nested factory methods" in str(e) + spec = Generator.generate("mockmod_dyn_typed.DynamicTyped obj") + assert spec["obj_value"] == expected_default + finally: + if "mockmod_dyn_typed" in sys.modules: + del sys.modules["mockmod_dyn_typed"] -def test_generate_with_typed_params() -> None: - """Test type-aware default generation.""" +def test_generate_multiple_typed_params() -> None: + """Test generating spec for a class with multiple typed parameters.""" - class TypedClass: - def __init__( - self, - int_param: int, - str_param: str, - float_param: float, - bool_param: bool, - opt_param: Optional[str] = None, - ) -> None: + class MultiTyped: + def __init__(self, year: int, name: str, ratio: float) -> None: pass class MockModule(ModuleType): def __init__(self) -> None: - super().__init__("mockmod_typed") - self.TypedClass = TypedClass + super().__init__("mockmod_multi_typed") + self.MultiTyped = MultiTyped mod = MockModule() - sys.modules["mockmod_typed"] = mod + sys.modules["mockmod_multi_typed"] = mod try: - spec = Generator.generate("mockmod_typed.TypedClass obj") + spec = Generator.generate("mockmod_multi_typed.MultiTyped obj") - # Use whole-output assertion with expected spec expected_spec: Spec = { - "mockmod_typed.TypedClass obj": { - "bool_param": "{obj_bool_param}", - "float_param": "{obj_float_param}", - "int_param": "{obj_int_param}", - "opt_param": "{obj_opt_param}", - "str_param": "{obj_str_param}", + "mockmod_multi_typed.MultiTyped obj": { + "name": "{obj_name}", + "ratio": "{obj_ratio}", + "year": "{obj_year}", }, - "obj_bool_param": False, - "obj_float_param": 0.0, - "obj_int_param": 0, - "obj_opt_param": None, - "obj_str_param": "", + "obj_name": "", + "obj_ratio": 0.0, + "obj_year": 0, } assert spec == expected_spec finally: - if "mockmod_typed" in sys.modules: - del sys.modules["mockmod_typed"] + if "mockmod_multi_typed" in sys.modules: + del sys.modules["mockmod_multi_typed"] def test_generate_with_dependency() -> None: @@ -320,44 +366,6 @@ def __init__(self) -> None: del sys.modules["mockmod_factory"] -def test_generate_list_and_dict_types() -> None: - """Test handling of list and dict type annotations.""" - from typing import Dict, List - - class Container: - def __init__( - self, - items: List[int], - mapping: Dict[str, int], - ) -> None: - pass - - class MockModule(ModuleType): - def __init__(self) -> None: - super().__init__("mockmod_container") - self.Container = Container - - mod = MockModule() - sys.modules["mockmod_container"] = mod - - try: - spec = Generator.generate("mockmod_container.Container c") - - # Use whole-output assertion with expected spec - expected_spec: Spec = { - "mockmod_container.Container c": { - "items": "{c_items}", - "mapping": "{c_mapping}", - }, - "c_items": [], - "c_mapping": {}, - } - assert spec == expected_spec - finally: - if "mockmod_container" in sys.modules: - del sys.modules["mockmod_container"] - - def test_generate_no_params() -> None: """Test generating spec for class with no parameters.""" @@ -526,129 +534,6 @@ def __init__(self) -> None: del sys.modules["mockmod_varargs"] -def test_generate_complex_union_type() -> None: - """Test handling of complex Union types (not just Optional).""" - from typing import Union - - class WithUnion: - def __init__(self, value: Union[int, str, float]) -> None: - pass - - class MockModule(ModuleType): - def __init__(self) -> None: - super().__init__("mockmod_union") - self.WithUnion = WithUnion - - mod = MockModule() - sys.modules["mockmod_union"] = mod - - try: - spec = Generator.generate("mockmod_union.WithUnion obj") - # Complex unions should default to None - expected_spec: Spec = { - "mockmod_union.WithUnion obj": { - "value": "{obj_value}", - }, - "obj_value": None, - } - assert spec == expected_spec - finally: - if "mockmod_union" in sys.modules: - del sys.modules["mockmod_union"] - - -def test_generate_tuple_type() -> None: - """Test handling of tuple type annotations.""" - from typing import Tuple - - class WithTuple: - def __init__(self, coords: Tuple[int, int]) -> None: - pass - - class MockModule(ModuleType): - def __init__(self) -> None: - super().__init__("mockmod_tuple") - self.WithTuple = WithTuple - - mod = MockModule() - sys.modules["mockmod_tuple"] = mod - - try: - spec = Generator.generate("mockmod_tuple.WithTuple obj") - expected_spec: Spec = { - "mockmod_tuple.WithTuple obj": { - "coords": "{obj_coords}", - }, - "obj_coords": (), # type: ignore[dict-item] - } - assert spec == expected_spec - finally: - if "mockmod_tuple" in sys.modules: - del sys.modules["mockmod_tuple"] - - -def test_generate_unknown_annotation() -> None: - """Test handling of unknown/complex annotation types.""" - - class WithCustom: - # Use a non-type annotation that's not in the standard set - def __init__(self, value) -> None: # type: ignore[no-untyped-def] - pass - - class MockModule(ModuleType): - def __init__(self) -> None: - super().__init__("mockmod_custom") - self.WithCustom = WithCustom # type: ignore[misc] - - mod = MockModule() - sys.modules["mockmod_custom"] = mod - - try: - spec = Generator.generate("mockmod_custom.WithCustom obj") - # Unknown types should default to None - expected_spec: Spec = { - "mockmod_custom.WithCustom obj": { - "value": "{obj_value}", - }, - "obj_value": None, - } - assert spec == expected_spec - finally: - if "mockmod_custom" in sys.modules: - del sys.modules["mockmod_custom"] - - -def test_generate_bytes_and_complex_types() -> None: - """Test handling of bytes and complex number types.""" - - class WithTypes: - def __init__(self, data: bytes, number: complex) -> None: - pass - - class MockModule(ModuleType): - def __init__(self) -> None: - super().__init__("mockmod_types") - self.WithTypes = WithTypes - - mod = MockModule() - sys.modules["mockmod_types"] = mod - - try: - spec = Generator.generate("mockmod_types.WithTypes obj") - expected_spec: Spec = { - "mockmod_types.WithTypes obj": { - "data": "{obj_data}", - "number": "{obj_number}", - }, - "obj_data": b"", - "obj_number": 0j, - } - assert spec == expected_spec - finally: - if "mockmod_types" in sys.modules: - del sys.modules["mockmod_types"] - - def test_generate_non_constant_default() -> None: """Test handling of non-constant defaults.""" diff --git a/tests/test_stdlib_compat.py b/tests/test_stdlib_compat.py index a6f8813..10a07bf 100644 --- a/tests/test_stdlib_compat.py +++ b/tests/test_stdlib_compat.py @@ -4,77 +4,54 @@ import datetime import pathlib -from typing import cast +from typing import Callable, cast -import apywire - - -def test_stdlib_int_list_args() -> None: - """Test instantiating int with positional arguments from a list.""" - spec: apywire.Spec = {"builtins.int myInt": [10]} - wired = apywire.Wiring(spec) - assert wired.myInt() == 10 +import pytest - -def test_stdlib_int_dict_numeric_args() -> None: - """Test instantiating int with positional arguments from numeric keys.""" - spec: apywire.Spec = {"builtins.int myInt": {0: 42}} - wired = apywire.Wiring(spec) - assert wired.myInt() == 42 +import apywire -def test_stdlib_complex_mixed_args() -> None: - """Test instantiating complex with mixed positional and keyword - arguments. +@pytest.mark.parametrize( + "class_path, args, expected", + [ + ("builtins.int", [10], 10), + ("builtins.int", {0: 42}, 42), + ("builtins.complex", {0: 1.5, "imag": 2.5}, complex(1.5, 2.5)), + ("pathlib.Path", ["/tmp", "foo"], pathlib.Path("/tmp/foo")), + ("datetime.date", [2023, 10, 27], datetime.date(2023, 10, 27)), + ], + ids=["int_list", "int_dict", "complex_mixed", "path", "date"], +) +def test_stdlib_instantiation( + class_path: str, args: object, expected: object +) -> None: + """Test instantiating various stdlib classes with different argument + formats. """ - # complex(real, imag) - # real as positional (0), imag as keyword - spec: apywire.Spec = {"builtins.complex myComplex": {0: 1.5, "imag": 2.5}} - wired = apywire.Wiring(spec) - c = cast(complex, wired.myComplex()) - assert c.real == 1.5 - assert c.imag == 2.5 - - -def test_stdlib_path_args() -> None: - """Test instantiating pathlib.Path with positional arguments.""" - spec: apywire.Spec = {"pathlib.Path myPath": ["/tmp", "foo"]} - wired = apywire.Wiring(spec) - p = wired.myPath() - assert isinstance(p, pathlib.Path) - assert str(p) == "/tmp/foo" - - -def test_stdlib_date_args() -> None: - """Test instantiating datetime.date with positional arguments.""" - spec: apywire.Spec = {"datetime.date myDate": [2023, 10, 27]} - wired = apywire.Wiring(spec) - d = wired.myDate() - assert d == datetime.date(2023, 10, 27) - - -def test_compile_list_args() -> None: - """Test compilation of list arguments.""" - spec: apywire.Spec = {"builtins.int myInt": [99]} - code = apywire.WiringCompiler(spec).compile() - - execd: dict[str, object] = {} - exec(code, execd) - compiled = execd["compiled"] - - # We can't easily type check the dynamic compiled object without a - # Protocol, but we can check the attribute. - assert cast(int, getattr(compiled, "myInt")()) == 99 - - -def test_compile_mixed_args() -> None: - """Test compilation of mixed numeric/string keys.""" - spec: apywire.Spec = {"builtins.complex myComplex": {0: 1.0, "imag": 2.0}} - code = apywire.WiringCompiler(spec).compile() + spec = {f"{class_path} obj": args} + wired = apywire.Wiring(spec) # type: ignore[arg-type] + assert wired.obj() == expected + + +@pytest.mark.parametrize( + "class_path, args, expected", + [ + ("builtins.int", [99], 99), + ("builtins.complex", {0: 1.0, "imag": 2.0}, complex(1.0, 2.0)), + ], + ids=["int", "complex"], +) +def test_compile_stdlib_instantiation( + class_path: str, args: object, expected: object +) -> None: + """Test compilation of various stdlib classes with different argument + formats. + """ + spec = {f"{class_path} obj": args} + code = apywire.WiringCompiler(spec).compile() # type: ignore[arg-type] execd: dict[str, object] = {} exec(code, execd) compiled = execd["compiled"] - c = cast(complex, getattr(compiled, "myComplex")()) - assert c == complex(1.0, 2.0) + assert cast(Callable[[], object], getattr(compiled, "obj"))() == expected