on_gmcp becomes public API, remove gmcp_log (jquast#127)

- .``_on_gmcp`` renamed to ``on_gmcp`` as a public API method
- ``gmcp_log`` isn't practical or useful for clients, removed

Author: jquast

docs/conf.py

@@ -22,7 +22,7 @@
     0, os.path.abspath(os.path.join(os.path.dirname(__file__), os.path.pardir))
 )
 
-suppress_warnings = ["image.nonlocal_uri"]
+suppress_warnings = ["image.nonlocal_uri", "ref.class"]
 
 autodoc_default_flags = [
     "members",

docs/history.rst

@@ -3,6 +3,8 @@ History
 3.0.1 (unreleased)
   * new: MCCP2 and MCCP3. Both client and server ends passively support if requested, and request
     support by --compression or deny support by --no-compression.
+  * new: MCCP2 and MCCP3. Both client and server ends passively support if requested, and request
+    support by --compression or deny support by --no-compression.
 
 3.0.0
   * change: :attr:`~telnetlib3.client_base.BaseClient.connect_minwait` default

telnetlib3/client.py

@@ -23,6 +23,8 @@
 __all__ = ("TelnetClient", "TelnetTerminalClient", "open_connection")
 
 #: Default GMCP modules requested via ``Core.Supports.Set``.
+#: Sub-modules are listed explicitly because not all servers treat
+#: top-level subscriptions as wildcards.
 _DEFAULT_GMCP_MODULES = [
     "Char 1",
     "Char.Vitals 1",
@@ -31,6 +33,7 @@
     "Room.Info 1",
     "Comm 1",
     "Comm.Channel 1",
+    "Group 1",
 ]
 
 
@@ -70,7 +73,6 @@ def __init__(
         waiter_closed: Optional[asyncio.Future[None]] = None,
         _waiter_connected: Optional[asyncio.Future[None]] = None,
         gmcp_modules: Optional[List[str]] = None,
-        gmcp_log: bool = False,
     ) -> None:
         """Initialize TelnetClient with terminal parameters."""
         self._compression = compression
@@ -86,7 +88,6 @@ def __init__(
             _waiter_connected=_waiter_connected,
         )
         self._gmcp_modules = gmcp_modules or list(_DEFAULT_GMCP_MODULES)
-        self._gmcp_log = gmcp_log
         self._gmcp_hello_sent = False
         self._send_environ = set(send_environ or self.DEFAULT_SEND_ENVIRON)
         self._extra.update(
@@ -166,7 +167,7 @@ def _setup_gmcp(self) -> None:
         """Wire GMCP callback and WILL-detection for Core.Hello handshake."""
         from telnetlib3.telopt import GMCP
 
-        self.writer.set_ext_callback(GMCP, self._on_gmcp)
+        self.writer.set_ext_callback(GMCP, self.on_gmcp)
 
         # Capture current handle_will (already includes CHARSET wrapper).
         # On reconnect, _original_handle_will was already restored in connection_made,
@@ -191,17 +192,14 @@ def _send_gmcp_hello(self) -> None:
         self.writer.send_gmcp("Core.Supports.Set", self._gmcp_modules)
         self.log.info("GMCP handshake: Core.Hello + Core.Supports.Set %s", self._gmcp_modules)
 
-    def _on_gmcp(self, package: str, data: Any) -> None:
+    def on_gmcp(self, package: str, data: Any) -> None:
         """Store incoming GMCP data on ``writer.ctx``, merging dict updates."""
         gmcp = self.writer.ctx.gmcp_data
         if isinstance(data, dict) and isinstance(gmcp.get(package), dict):
             gmcp[package].update(data)
         else:
             gmcp[package] = data
-        if self._gmcp_log:
-            self.log.info("GMCP: %s %r", package, data)
-        else:
-            self.log.debug("GMCP: %s %r", package, data)
+        self.log.debug("GMCP: %s %r", package, data)
 
     def send_ttype(self) -> str:
         """Callback for responding to TTYPE requests."""
@@ -628,12 +626,10 @@ async def run_client() -> None:
     # flags before negotiation starts.
     encoding_explicit = args["encoding"] not in ("utf8", "utf-8", False)
     gmcp_modules: Optional[List[str]] = args.get("gmcp_modules")
-    gmcp_log: bool = args.get("gmcp_log", False)
 
     def _client_factory(**kwargs: Any) -> client_base.BaseClient:
         client: TelnetClient
         kwargs["gmcp_modules"] = gmcp_modules
-        kwargs["gmcp_log"] = gmcp_log
         if sys.platform != "win32" and sys.stdin.isatty():
             client = TelnetTerminalClient(**kwargs)
         else:
@@ -940,12 +936,6 @@ def _get_argument_parser() -> argparse.ArgumentParser:
         '(e.g. "Char 1,Room 1,IRE.Rift 1"). '
         "When provided, replaces the built-in defaults.",
     )
-    parser.add_argument(
-        "--gmcp-log",
-        action="store_true",
-        default=False,
-        help="log all incoming GMCP messages at INFO level " "(default: DEBUG only)",
-    )
     parser.add_argument(
         "--typescript",
         default=None,
@@ -1047,7 +1037,6 @@ def _transform_args(args: argparse.Namespace) -> Dict[str, Any]:
             else None
         ),
         "compression": args.compression,
-        "gmcp_log": args.gmcp_log,
         "typescript": args.typescript,
     }
 

telnetlib3/client_base.py

@@ -478,9 +478,7 @@ def _mccp3_end(self) -> None:
         """Stop MCCP3 compression, flush Z_FINISH."""
         if self._mccp3_compressor is not None:
             if not self.writer.is_closing():
-                self._mccp3_orig_write(
-                    self._mccp3_compressor.flush(zlib.Z_FINISH)
-                )
+                self._mccp3_orig_write(self._mccp3_compressor.flush(zlib.Z_FINISH))
             self._mccp3_compressor = None
             # Restore original transport.write
             self.writer._transport.write = self._mccp3_orig_write  # type: ignore[method-assign]

telnetlib3/server.py

@@ -217,9 +217,7 @@ def _mccp2_end(self) -> None:
         """Stop MCCP2 compression, flush Z_FINISH."""
         if self._mccp2_compressor is not None:
             try:
-                self._mccp2_orig_write(
-                    self._mccp2_compressor.flush(zlib.Z_FINISH)
-                )
+                self._mccp2_orig_write(self._mccp2_compressor.flush(zlib.Z_FINISH))
             except zlib.error as exc:
                 logger.debug("MCCP2 Z_FINISH flush error: %s", exc)
             self._mccp2_compressor = None

telnetlib3/stream_writer.py

@@ -3047,8 +3047,7 @@ def _handle_sb_mccp3(self, buf: collections.deque[bytes]) -> None:
         """
         Handle MCCP3 subnegotiation (``IAC SB MCCP3 IAC SE``).
 
-        On server side, this signals that subsequent client→server data is
-        zlib-compressed.
+        On server side, this signals that subsequent client→server data is zlib-compressed.
 
         :param buf: bytes following IAC SB MCCP3_COMPRESS.
         """

telnetlib3/tests/test_client_unit.py

@@ -557,15 +557,15 @@ def _make_connected_client(**kwargs):
 @pytest.mark.asyncio
 async def test_on_gmcp_stores_on_writer_ctx():
     client, _ = _make_connected_client()
-    client._on_gmcp("Room.Info", {"name": "Town Square"})
+    client.on_gmcp("Room.Info", {"name": "Town Square"})
     assert client.writer.ctx.gmcp_data["Room.Info"] == {"name": "Town Square"}
 
 
 @pytest.mark.asyncio
 async def test_on_gmcp_merges_dicts_on_writer_ctx():
     client, _ = _make_connected_client()
-    client._on_gmcp("Char.Vitals", {"hp": 100, "maxhp": 100})
-    client._on_gmcp("Char.Vitals", {"hp": 63})
+    client.on_gmcp("Char.Vitals", {"hp": 100, "maxhp": 100})
+    client.on_gmcp("Char.Vitals", {"hp": 63})
     assert client.writer.ctx.gmcp_data["Char.Vitals"] == {"hp": 63, "maxhp": 100}
 
 

telnetlib3/tests/test_mccp.py

@@ -9,17 +9,7 @@
 import pytest
 
 # local
-from telnetlib3.telopt import (
-    DO,
-    SB,
-    SE,
-    IAC,
-    DONT,
-    WILL,
-    WONT,
-    MCCP2_COMPRESS,
-    MCCP3_COMPRESS,
-)
+from telnetlib3.telopt import DO, SB, SE, IAC, DONT, WILL, WONT, MCCP2_COMPRESS, MCCP3_COMPRESS
 from telnetlib3.stream_writer import TelnetWriter
 from telnetlib3.tests.accessories import MockProtocol, MockTransport
 
@@ -184,9 +174,7 @@ def test_mid_chunk_split(self):
         compressed += compressor.flush(zlib.Z_SYNC_FLUSH)
 
         chunk = IAC + SB + MCCP2_COMPRESS + IAC + SE + compressed
-        cmd_received = _process_data_chunk(
-            chunk, w, reader, None, lambda *a: None
-        )
+        cmd_received = _process_data_chunk(chunk, w, reader, None, lambda *a: None)
         assert cmd_received is True
         assert w.mccp2_active is True
         assert w._compressed_remainder == compressed
@@ -200,9 +188,7 @@ def test_no_remainder_without_mccp2(self):
         w, _t, _p = new_writer(server=False, client=True, reader=reader)
 
         chunk = b"plain text data"
-        cmd_received = _process_data_chunk(
-            chunk, w, reader, None, lambda *a: None
-        )
+        cmd_received = _process_data_chunk(chunk, w, reader, None, lambda *a: None)
         assert cmd_received is False
         assert w._compressed_remainder is None
 
@@ -249,11 +235,7 @@ def test_initial_attributes(self):
         assert w.mccp2_active is False
         assert w.mccp3_active is False
 
-    @pytest.mark.parametrize(
-        "opt",
-        [MCCP2_COMPRESS, MCCP3_COMPRESS],
-        ids=["MCCP2", "MCCP3"],
-    )
+    @pytest.mark.parametrize("opt", [MCCP2_COMPRESS, MCCP3_COMPRESS], ids=["MCCP2", "MCCP3"])
     def test_empty_sb_allowed(self, opt):
         w, _t, _p = new_writer(server=False, client=True)
         w.pending_option[SB + opt] = True
@@ -275,7 +257,10 @@ def _make_compressed(plaintext: bytes, finish: bool = False) -> bytes:
 _BOUNDARY_SB = IAC + SB + MCCP2_COMPRESS + IAC + SE
 _BOUNDARY_FULL = _BOUNDARY_SB + _BOUNDARY_COMPRESSED
 _BOUNDARY_SPLITS = [
-    1, 2, 3, 4,
+    1,
+    2,
+    3,
+    4,
     len(_BOUNDARY_SB) - 1,
     len(_BOUNDARY_SB),
     len(_BOUNDARY_SB) + 1,
@@ -317,11 +302,7 @@ async def test_two_chunk_delivery(self, split_at):
         joined = b"".join(received)
         assert joined == _BOUNDARY_PLAINTEXT
 
-    @pytest.mark.parametrize(
-        "n_chunks",
-        [3, 5, 10],
-        ids=["3_chunks", "5_chunks", "10_chunks"],
-    )
+    @pytest.mark.parametrize("n_chunks", [3, 5, 10], ids=["3_chunks", "5_chunks", "10_chunks"])
     async def test_multi_chunk_delivery(self, n_chunks):
         """Compressed data delivered in many small chunks."""
         client, received = _make_client_with_capture()
@@ -347,9 +328,7 @@ async def test_z_finish_with_trailing_plaintext(self):
         assert joined == plaintext + trailing
 
     @pytest.mark.parametrize(
-        "split_at",
-        [1, 4, 8, 16],
-        ids=["byte_1", "byte_4", "byte_8", "byte_16"],
+        "split_at", [1, 4, 8, 16], ids=["byte_1", "byte_4", "byte_8", "byte_16"]
     )
     async def test_compressed_only_boundary(self, split_at):
         """Split within compressed data only (SB already processed)."""