AsyncAPI 3.0.0 Redis publisher tests

Author: KrySeyt

tests/asyncapi/redis/v3_0_0/test_publisher.py

@@ -0,0 +1,50 @@
+from faststream.asyncapi.generate import get_app_schema
+from faststream.redis import RedisBroker
+from tests.asyncapi.base.v3_0_0.publisher import PublisherTestcase
+
+
+class TestArguments(PublisherTestcase):
+    broker_class = RedisBroker
+
+    def test_channel_publisher(self):
+        broker = self.broker_class()
+
+        @broker.publisher("test")
+        async def handle(msg): ...
+
+        schema = get_app_schema(self.build_app(broker)).to_jsonable()
+        key = tuple(schema["channels"].keys())[0]  # noqa: RUF015
+
+        assert schema["channels"][key]["bindings"] == {
+            "redis": {
+                "bindingVersion": "custom",
+                "channel": "test",
+                "method": "publish",
+            }
+        }
+
+    def test_list_publisher(self):
+        broker = self.broker_class()
+
+        @broker.publisher(list="test")
+        async def handle(msg): ...
+
+        schema = get_app_schema(self.build_app(broker)).to_jsonable()
+        key = tuple(schema["channels"].keys())[0]  # noqa: RUF015
+
+        assert schema["channels"][key]["bindings"] == {
+            "redis": {"bindingVersion": "custom", "channel": "test", "method": "rpush"}
+        }
+
+    def test_stream_publisher(self):
+        broker = self.broker_class()
+
+        @broker.publisher(stream="test")
+        async def handle(msg): ...
+
+        schema = get_app_schema(self.build_app(broker)).to_jsonable()
+        key = tuple(schema["channels"].keys())[0]  # noqa: RUF015
+
+        assert schema["channels"][key]["bindings"] == {
+            "redis": {"bindingVersion": "custom", "channel": "test", "method": "xadd"}
+        }