test: cover channels consumer

Author: syrusakbary

src/tests/channels/test_consumer.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from graphql_server.channels.handlers.base import ChannelsConsumer
+
+
+class DummyChannelLayer:
+    def __init__(self) -> None:
+        self.added: list[tuple[str, str]] = []
+        self.discarded: list[tuple[str, str]] = []
+
+    async def group_add(self, group: str, channel: str) -> None:
+        self.added.append((group, channel))
+
+    async def group_discard(self, group: str, channel: str) -> None:
+        self.discarded.append((group, channel))
+
+
+@pytest.mark.asyncio
+async def test_channel_listen_receives_messages_and_cleans_up() -> None:
+    consumer = ChannelsConsumer()
+    layer = DummyChannelLayer()
+    consumer.channel_layer = layer
+    consumer.channel_name = "chan"
+
+    gen = consumer.channel_listen("test.message", groups=["g"], timeout=0.1)
+
+    async def send() -> None:
+        await asyncio.sleep(0)
+        queue = next(iter(consumer.listen_queues["test.message"]))
+        queue.put_nowait({"type": "test.message", "payload": 1})
+
+    asyncio.create_task(send())
+
+    with pytest.deprecated_call(match="Use listen_to_channel instead"):
+        message = await gen.__anext__()
+    assert message == {"type": "test.message", "payload": 1}
+
+    await gen.aclose()
+
+    assert layer.added == [("g", "chan")]
+    assert layer.discarded == [("g", "chan")]
+
+
+@pytest.mark.asyncio
+async def test_channel_listen_times_out() -> None:
+    consumer = ChannelsConsumer()
+    layer = DummyChannelLayer()
+    consumer.channel_layer = layer
+    consumer.channel_name = "chan"
+
+    gen = consumer.channel_listen("test.message", groups=["g"], timeout=0.01)
+
+    with pytest.deprecated_call(match="Use listen_to_channel instead"):
+        with pytest.raises(StopAsyncIteration):
+            await gen.__anext__()
+
+    assert layer.added == [("g", "chan")]
+    assert layer.discarded == [("g", "chan")]

src/tests/utils/test_debug.py

@@ -0,0 +1,46 @@
+import builtins
+
+import pytest
+
+from graphql_server.utils.debug import GraphQLJSONEncoder, pretty_print_graphql_operation
+
+
+def test_graphql_json_encoder_default():
+    class Foo:
+        pass
+
+    foo = Foo()
+    encoder = GraphQLJSONEncoder()
+    assert encoder.default(foo) == repr(foo)
+
+
+def test_pretty_print_requires_pygments(monkeypatch):
+    original_import = builtins.__import__
+
+    def fake_import(name, *args, **kwargs):
+        if name.startswith("pygments"):
+            raise ImportError("No module named pygments")
+        return original_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "__import__", fake_import)
+    with pytest.raises(ImportError):
+        pretty_print_graphql_operation("Query", "query { field }", None)
+
+
+def test_pretty_print_graphql_operation(capsys):
+    obj = object()
+    variables = {"var": obj}
+    pretty_print_graphql_operation("MyQuery", "query { field }", variables)
+    captured = capsys.readouterr().out
+    assert "MyQuery" in captured
+    assert "field" in captured
+    assert "var" in captured
+    assert repr(obj) in captured
+
+
+def test_pretty_print_introspection_query(capsys):
+    pretty_print_graphql_operation(
+        "IntrospectionQuery", "query { __schema { queryType { name } } }", None
+    )
+    captured = capsys.readouterr().out
+    assert captured == ""