test_input_special_chars.py

from io import StringIO

import pytest

from replit_river.codegen.client import schema_to_river_client_codegen


def test_input_special_chars_basemodel() -> None:
    """Handles special characters in input field names for BaseModel."""

    schema_to_river_client_codegen(
        read_schema=lambda: open(
            "tests/v1/codegen/rpc/input-special-chars-schema.json"
        ),  # noqa: E501
        target_path="tests/v1/codegen/rpc/generated_input_special",
        client_name="InputSpecialClient",
        typed_dict_inputs=False,  # BaseModel inputs
        file_opener=lambda _: StringIO(),
        method_filter=None,
        protocol_version="v1.1",
    )


def test_input_special_chars_typeddict() -> None:
    """Handles special characters in input field names for TypedDict."""

    # Test should pass without raising an exception
    schema_to_river_client_codegen(
        read_schema=lambda: open(
            "tests/v1/codegen/rpc/input-special-chars-schema.json"
        ),  # noqa: E501
        target_path="tests/v1/codegen/rpc/generated_input_special_td",
        client_name="InputSpecialTDClient",
        typed_dict_inputs=True,  # TypedDict inputs
        file_opener=lambda _: StringIO(),
        method_filter=None,
        protocol_version="v1.1",
    )


def test_input_collision_error_basemodel() -> None:
    """Raises ValueError for input field name collisions with BaseModel."""

    with pytest.raises(ValueError) as exc_info:
        schema_to_river_client_codegen(
            read_schema=lambda: open(
                "tests/v1/codegen/rpc/input-collision-schema.json"
            ),  # noqa: E501
            target_path="tests/v1/codegen/rpc/generated_input_collision",
            client_name="InputCollisionClient",
            typed_dict_inputs=False,  # BaseModel inputs
            file_opener=lambda _: StringIO(),
            method_filter=None,
            protocol_version="v1.1",
        )

    # Check that the error message matches the expected format for field name collision
    error_message = str(exc_info.value)
    assert "Field name collision" in error_message
    assert "data-3" in error_message
    assert "data:3" in error_message
    assert "all normalize to the same effective name 'data_3'" in error_message


def test_input_collision_error_typeddict() -> None:
    """Raises ValueError for input field name collisions with TypedDict."""

    with pytest.raises(ValueError) as exc_info:
        schema_to_river_client_codegen(
            read_schema=lambda: open(
                "tests/v1/codegen/rpc/input-collision-schema.json"
            ),  # noqa: E501
            target_path="tests/v1/codegen/rpc/generated_input_collision_td",
            client_name="InputCollisionTDClient",
            typed_dict_inputs=True,  # TypedDict inputs
            file_opener=lambda _: StringIO(),
            method_filter=None,
            protocol_version="v1.1",
        )

    # Check that the error message matches the expected format for field name collision
    error_message = str(exc_info.value)
    assert "Field name collision" in error_message
    assert "data-3" in error_message
    assert "data:3" in error_message
    assert "all normalize to the same effective name 'data_3'" in error_message


def test_init_special_chars_basemodel() -> None:
    """Handles special characters in init field names for BaseModel."""

    init_schema = {
        "services": {
            "test_service": {
                "procedures": {
                    "stream_method": {
                        "init": {
                            "type": "object",
                            "properties": {
                                "init-field1": {"type": "string"},
                                "init:field2": {"type": "number"},
                                "init.field3": {"type": "boolean"},
                            },
                            "required": ["init-field1"],
                        },
                        "output": {"type": "boolean"},
                        "errors": {"not": {}},
                        "type": "stream",
                    }
                }
            }
        }
    }

    import json

    # Test should pass without raising an exception
    schema_to_river_client_codegen(
        read_schema=lambda: StringIO(json.dumps(init_schema)),
        target_path="tests/v1/codegen/rpc/generated_init_special",
        client_name="InitSpecialClient",
        typed_dict_inputs=False,  # BaseModel inputs
        file_opener=lambda _: StringIO(),
        method_filter=None,
        protocol_version="v2.0",
    )


def test_init_special_chars_typeddict() -> None:
    """Handles special characters in init field names for TypedDict."""

    init_schema = {
        "services": {
            "test_service": {
                "procedures": {
                    "stream_method": {
                        "init": {
                            "type": "object",
                            "properties": {
                                "init-field1": {"type": "string"},
                                "init:field2": {"type": "number"},
                                "init.field3": {"type": "boolean"},
                            },
                            "required": ["init-field1"],
                        },
                        "output": {"type": "boolean"},
                        "errors": {"not": {}},
                        "type": "stream",
                    }
                }
            }
        }
    }

    import json

    # Test should pass without raising an exception
    schema_to_river_client_codegen(
        read_schema=lambda: StringIO(json.dumps(init_schema)),
        target_path="tests/v1/codegen/rpc/generated_init_special_td",
        client_name="InitSpecialTDClient",
        typed_dict_inputs=True,  # TypedDict inputs
        file_opener=lambda _: StringIO(),
        method_filter=None,
        protocol_version="v2.0",
    )


class UnclosableStringIO(StringIO):
    def close(self) -> None:
        pass


def test_python_keyword_field_names_basemodel() -> None:
    """Handles Python reserved keywords as field names for BaseModel."""

    import ast
    import json
    from pathlib import Path

    keyword_schema = {
        "services": {
            "test_service": {
                "procedures": {
                    "rpc_method": {
                        "input": {
                            "type": "object",
                            "properties": {
                                "from": {"type": "string"},
                                "to": {"type": "string"},
                                "class": {"type": "number"},
                                "import": {"type": "boolean"},
                            },
                            "required": ["from", "to"],
                        },
                        "output": {
                            "type": "object",
                            "properties": {
                                "from": {"type": "string"},
                                "to": {"type": "string"},
                            },
                            "required": ["from", "to"],
                        },
                        "errors": {"not": {}},
                        "type": "rpc",
                    }
                }
            }
        }
    }

    files: dict[Path, UnclosableStringIO] = {}

    def file_opener(path: Path) -> UnclosableStringIO:
        buf = UnclosableStringIO()
        files[path] = buf
        return buf

    schema_to_river_client_codegen(
        read_schema=lambda: StringIO(json.dumps(keyword_schema)),
        target_path="test_keyword_bm",
        client_name="KeywordBMClient",
        typed_dict_inputs=False,
        file_opener=file_opener,
        method_filter=None,
        protocol_version="v1.1",
    )

    # Verify all generated files are valid Python
    for path, buf in files.items():
        buf.seek(0)
        content = buf.read()
        try:
            ast.parse(content)
        except SyntaxError as e:
            raise AssertionError(
                f"Generated file {path} has invalid syntax: {e}\n{content}"
            )


def test_python_keyword_field_names_typeddict() -> None:
    """Handles Python reserved keywords as field names for TypedDict."""

    import ast
    import json
    from pathlib import Path

    keyword_schema = {
        "services": {
            "test_service": {
                "procedures": {
                    "rpc_method": {
                        "input": {
                            "type": "object",
                            "properties": {
                                "from": {"type": "string"},
                                "to": {"type": "string"},
                                "class": {"type": "number"},
                                "import": {"type": "boolean"},
                            },
                            "required": ["from", "to"],
                        },
                        "output": {
                            "type": "object",
                            "properties": {
                                "from": {"type": "string"},
                                "to": {"type": "string"},
                            },
                            "required": ["from", "to"],
                        },
                        "errors": {"not": {}},
                        "type": "rpc",
                    }
                }
            }
        }
    }

    files: dict[Path, UnclosableStringIO] = {}

    def file_opener(path: Path) -> UnclosableStringIO:
        buf = UnclosableStringIO()
        files[path] = buf
        return buf

    schema_to_river_client_codegen(
        read_schema=lambda: StringIO(json.dumps(keyword_schema)),
        target_path="test_keyword_td",
        client_name="KeywordTDClient",
        typed_dict_inputs=True,
        file_opener=file_opener,
        method_filter=None,
        protocol_version="v1.1",
    )

    # Verify all generated files are valid Python
    for path, buf in files.items():
        buf.seek(0)
        content = buf.read()
        try:
            ast.parse(content)
        except SyntaxError as e:
            raise AssertionError(
                f"Generated file {path} has invalid syntax: {e}\n{content}"
            )