Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
436 changes: 224 additions & 212 deletions apywire/compiler.py

Large diffs are not rendered by default.

4 changes: 4 additions & 0 deletions apywire/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,13 +8,17 @@
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
PLACEHOLDER_END = "}" # End marker for placeholder references

# 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

Expand Down
4 changes: 2 additions & 2 deletions apywire/runtime.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down
7 changes: 3 additions & 4 deletions apywire/wiring.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@

from apywire.constants import (
PLACEHOLDER_END,
PLACEHOLDER_PATTERN,
PLACEHOLDER_REGEX,
PLACEHOLDER_START,
SPEC_KEY_DELIMITER,
SYNTHETIC_CONST,
Expand Down Expand Up @@ -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] = {}
Expand Down Expand Up @@ -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.
Expand All @@ -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():
Expand Down
200 changes: 55 additions & 145 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand All @@ -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()
Expand Down Expand Up @@ -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:
Expand All @@ -225,43 +168,27 @@ 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()
assert "class Compiled:" in output
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": {}}'
Expand Down Expand Up @@ -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:
Expand Down
41 changes: 17 additions & 24 deletions tests/test_compile_aio.py
Original file line number Diff line number Diff line change
Expand Up @@ -250,38 +250,31 @@ 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)
compiled = execd["compiled"]

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())

Expand Down
Loading