Fix producer parameter validation (#33)

* Fix edge case

* Add tests

* Move parameter count validation from amqp to receiver plane

* Increment version

Author: WiZeYAR

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "asyncapi-python"
-version = "0.3.0rc6"
+version = "0.3.0rc8"
 license = { text = "Apache-2.0" }
 description = "Easily generate type-safe and async Python applications from AsyncAPI 3 specifications."
 authors = [{ name = "Yaroslav Petrov", email = "[email protected]" }]

src/asyncapi_python/contrib/wire/amqp/resolver.py

@@ -6,10 +6,7 @@
 from asyncapi_python.kernel.document.bindings import AmqpChannelBinding
 from asyncapi_python.kernel.document.channel import Channel
 from asyncapi_python.kernel.wire import EndpointParams
-from asyncapi_python.kernel.wire.utils import (
-    substitute_parameters,
-    validate_parameters_strict,
-)
+from asyncapi_python.kernel.wire.utils import substitute_parameters
 
 from .config import AmqpBindingType, AmqpConfig
 
@@ -186,8 +183,7 @@ def resolve_amqp_config(
 
         # Channel address pattern (with parameter substitution)
         case (False, None, address, _) if address:
-            # Strict validation for implicit queue binding
-            validate_parameters_strict(channel, param_values)
+            # Validate no wildcards for implicit queue binding
             _validate_no_wildcards_in_queue(param_values)
             resolved_address = substitute_parameters(address, param_values)
             return AmqpConfig(
@@ -200,8 +196,7 @@ def resolve_amqp_config(
 
         # Operation name pattern (fallback)
         case (False, None, None, op_name) if op_name:
-            # Strict validation for implicit queue binding
-            validate_parameters_strict(channel, param_values)
+            # Validate no wildcards for implicit queue binding
             _validate_no_wildcards_in_queue(param_values)
             return AmqpConfig(
                 queue_name=op_name,
@@ -228,13 +223,9 @@ def resolve_queue_binding(
     """Resolve AMQP queue binding configuration
 
     Queue bindings require:
-    - All channel parameters must be provided (strict validation)
     - No wildcards allowed in parameter values
     """
 
-    # Strict validation: all parameters required, exact match
-    validate_parameters_strict(channel, param_values)
-
     # Validate no wildcards in queue binding parameters
     _validate_no_wildcards_in_queue(param_values)
 
@@ -280,14 +271,10 @@ def resolve_routing_key_binding(
     """Resolve AMQP routing key binding configuration for pub/sub patterns
 
     For routing key bindings:
-    - All channel-defined parameters must be provided (strict validation)
     - Parameter values can explicitly contain wildcards ('*' or '#')
     - Wildcards are allowed for topic exchange pattern matching
     """
 
-    # Strict validation: all parameters required, exact match
-    validate_parameters_strict(channel, param_values)
-
     # Determine exchange name and type
     # For exchange name, we need concrete values (no wildcards)
     # If param_values has placeholders, use them; otherwise use literal exchange name

src/asyncapi_python/kernel/endpoint/rpc_server.py

@@ -4,6 +4,7 @@
 from typing_extensions import Unpack
 
 from asyncapi_python.kernel.wire import Consumer, Producer
+from asyncapi_python.kernel.wire.utils import validate_parameters_strict
 
 from ..exceptions import Reject
 from ..typing import (
@@ -63,6 +64,11 @@ async def start(self, **params: Unpack[AbstractEndpoint.StartParams]) -> None:
         if not self._reply_codecs:
             raise RuntimeError("RPC server operation has no reply messages defined")
 
+        # Validate subscription parameters before creating consumer
+        validate_parameters_strict(
+            self._operation.channel, self._subscription_parameters
+        )
+
         # Create consumer for receiving requests
         self._consumer = await self._wire.create_consumer(
             channel=self._operation.channel,

src/asyncapi_python/kernel/endpoint/subscriber.py

@@ -4,6 +4,7 @@
 from typing_extensions import Unpack
 
 from asyncapi_python.kernel.wire import Consumer
+from asyncapi_python.kernel.wire.utils import validate_parameters_strict
 
 from ..exceptions import Reject
 from ..typing import BatchConfig, BatchConsumer, Handler, IncomingMessage, T_Input
@@ -46,6 +47,11 @@ async def start(self, **params: Unpack[AbstractEndpoint.StartParams]) -> None:
                 f"Use @{self._operation.key} decorator to register a handler function."
             )
 
+        # Validate subscription parameters before creating consumer
+        validate_parameters_strict(
+            self._operation.channel, self._subscription_parameters
+        )
+
         # Create consumer from wire factory
         self._consumer = await self._wire.create_consumer(
             channel=self._operation.channel,
@@ -331,3 +337,4 @@ async def process_batch():
                     # If processing remaining batch fails, just nack all and continue
                     for _, wire_message in batch:
                         await wire_message.nack()
+                        await wire_message.nack()

tests/core/wire/test_amqp_resolver.py

@@ -196,157 +196,3 @@ def test_resolve_channel_address_without_binding_accepts_concrete_params():
 
     # Verify the address was substituted with concrete value
     assert config.queue_name == "task.high"
-
-
-def test_resolve_amqp_config_rejects_missing_parameters():
-    """Resolver should validate all required parameters are provided."""
-    from asyncapi_python.kernel.document.channel import AddressParameter
-
-    # Create channel with 2 required parameters
-    channel = create_test_channel(
-        address="weather.{location}.{severity}",
-        binding=AmqpChannelBinding(type="queue"),
-    )
-    # Add parameter definitions to channel
-    channel = Channel(
-        key=channel.key,
-        address=channel.address,
-        title=channel.title,
-        summary=channel.summary,
-        description=channel.description,
-        servers=channel.servers,
-        messages=channel.messages,
-        parameters={
-            "location": AddressParameter(
-                key="location",
-                description="Geographic location",
-                location=None,
-            ),
-            "severity": AddressParameter(
-                key="severity",
-                description="Alert severity",
-                location=None,
-            ),
-        },
-        tags=channel.tags,
-        external_docs=channel.external_docs,
-        bindings=channel.bindings,
-    )
-
-    params: EndpointParams = {
-        "channel": channel,
-        "parameters": {"location": "NYC"},  # Missing severity
-        "op_bindings": None,
-        "is_reply": False,
-    }
-
-    with pytest.raises(ValueError) as exc_info:
-        resolve_amqp_config(params, "test_op", "test_app")
-
-    error_msg = str(exc_info.value)
-    assert "Missing required parameters" in error_msg
-    assert "severity" in error_msg
-
-
-def test_resolve_amqp_config_rejects_extra_parameters():
-    """Resolver should reject extra parameters not defined in channel."""
-    from asyncapi_python.kernel.document.channel import AddressParameter
-
-    # Create channel with 1 required parameter
-    channel = create_test_channel(
-        address="weather.{location}",
-        binding=AmqpChannelBinding(type="queue"),
-    )
-    # Add parameter definition to channel
-    channel = Channel(
-        key=channel.key,
-        address=channel.address,
-        title=channel.title,
-        summary=channel.summary,
-        description=channel.description,
-        servers=channel.servers,
-        messages=channel.messages,
-        parameters={
-            "location": AddressParameter(
-                key="location",
-                description="Geographic location",
-                location=None,
-            ),
-        },
-        tags=channel.tags,
-        external_docs=channel.external_docs,
-        bindings=channel.bindings,
-    )
-
-    params: EndpointParams = {
-        "channel": channel,
-        "parameters": {
-            "location": "NYC",
-            "severity": "high",  # Extra parameter
-        },
-        "op_bindings": None,
-        "is_reply": False,
-    }
-
-    with pytest.raises(ValueError) as exc_info:
-        resolve_amqp_config(params, "test_op", "test_app")
-
-    error_msg = str(exc_info.value)
-    assert "Unexpected parameters" in error_msg
-    assert "severity" in error_msg
-
-
-def test_resolve_amqp_config_routing_key_rejects_missing_parameters():
-    """Routing key bindings should also validate all parameters are provided."""
-    from asyncapi_python.kernel.document.channel import AddressParameter
-
-    # Create channel with routingKey binding
-    channel = create_test_channel(
-        address="weather.{location}.{severity}",
-        binding=AmqpChannelBinding(
-            type="routingKey",
-            exchange=AmqpExchange(
-                name="weather_exchange",
-                type=AmqpExchangeType.TOPIC,
-            ),
-        ),
-    )
-    # Add parameter definitions
-    channel = Channel(
-        key=channel.key,
-        address=channel.address,
-        title=channel.title,
-        summary=channel.summary,
-        description=channel.description,
-        servers=channel.servers,
-        messages=channel.messages,
-        parameters={
-            "location": AddressParameter(
-                key="location",
-                description="Geographic location",
-                location=None,
-            ),
-            "severity": AddressParameter(
-                key="severity",
-                description="Alert severity",
-                location=None,
-            ),
-        },
-        tags=channel.tags,
-        external_docs=channel.external_docs,
-        bindings=channel.bindings,
-    )
-
-    params: EndpointParams = {
-        "channel": channel,
-        "parameters": {"location": "NYC"},  # Missing severity
-        "op_bindings": None,
-        "is_reply": False,
-    }
-
-    with pytest.raises(ValueError) as exc_info:
-        resolve_amqp_config(params, "test_op", "test_app")
-
-    error_msg = str(exc_info.value)
-    assert "Missing required parameters" in error_msg
-    assert "severity" in error_msg