[pull] master from jquast:master

docs/history.rst

@@ -1,6 +1,11 @@
 History
 =======
 
+5.0.1
+  * bugfix: :func:`~telnetlib3.client.open_connection` raised ``AttributeError: 'NoneType' object
+    has no attribute 'isatty'`` in processes without a console, like ``pythonw.exe`` on Windows.
+
+
 5.0.0
   * changed: :meth:`~telnetlib3.stream_writer.TelnetWriter.handle_zmp` now receives ``command,
     *args`` instead of one ``parts`` list; ``zmp_data`` moved to ``writer.ctx`` and is now a dict

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "telnetlib3"
-version = "5.0.0"  # Keep in sync with telnetlib3/accessories.py::get_version !
+version = "5.0.1"  # Keep in sync with telnetlib3/accessories.py::get_version !
 description = " Python Telnet server and client CLI and Protocol library"
 readme = "README.rst"
 license = "ISC"

telnetlib3/accessories.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 # std imports
+import os
+import sys
 import shlex
 import asyncio
 import logging
@@ -26,6 +28,7 @@
     "repr_mapping",
     "function_lookup",
     "make_reader_task",
+    "is_a_tty",
 )
 
 PATIENCE_MESSAGES = [
@@ -42,7 +45,7 @@
 
 def get_version() -> str:
     """Return the current version of telnetlib3."""
-    return "5.0.0"  # keep in sync with pyproject.toml !
+    return "5.0.1"  # keep in sync with pyproject.toml !
 
 
 def encoding_from_lang(lang: str) -> Optional[str]:
@@ -172,3 +175,32 @@ def make_reader_task(
 ) -> "asyncio.Task[Any]":
     """Return asyncio task wrapping coroutine of reader.read(size)."""
     return asyncio.ensure_future(reader.read(size))
+
+
+def is_a_tty() -> bool:
+    """
+    Whether an interactive terminal is available on standard input.
+
+    This is a more conservative test than ``sys.stdin.isatty()``: the standard
+    streams may be ``None`` (running under ``pythonw.exe`` on Windows, or any
+    process launched without a console), replaced by an object without a
+    working :meth:`~io.IOBase.isatty` or :meth:`~io.IOBase.fileno` (test
+    harnesses, GUI frameworks), or detached.  A real file descriptor is
+    required, because terminal features such as window size and raw mode are
+    performed against ``sys.stdin.fileno()``.
+
+    :returns: ``True`` only when stdin is a terminal backed by a file
+        descriptor.
+    """
+    stdin = sys.stdin
+    if stdin is None:
+        return False
+    try:
+        if not stdin.isatty():
+            return False
+        return os.isatty(stdin.fileno())
+    except (AttributeError, ValueError, OSError):
+        # AttributeError: stream without isatty/fileno,
+        # ValueError: closed or detached stream,
+        # OSError: fileno() unsupported by the underlying object.
+        return False

telnetlib3/client.py

@@ -577,7 +577,9 @@ async def open_connection(
     :param port: Remote Internet host TCP port.
     :param client_factory: Client connection class factory.  When ``None``,
         :class:`TelnetTerminalClient` is used when *stdin* is attached to a
-        terminal, :class:`TelnetClient` otherwise.
+        terminal, :class:`TelnetClient` otherwise.  Processes without a
+        console, such as those launched by ``pythonw.exe`` on Windows, where
+        ``sys.stdin`` is ``None``, always receive :class:`TelnetClient`.
     :param family: Same meaning as
         :meth:`asyncio.loop.create_connection`.
     :param flags: Same meaning as
@@ -643,7 +645,7 @@ async def open_connection(
     """
     if client_factory is None:
         client_factory = TelnetClient
-        if sys.stdin.isatty():
+        if accessories.is_a_tty():
             client_factory = TelnetTerminalClient
 
     def connection_factory() -> client_base.BaseClient:
@@ -724,7 +726,7 @@ async def run_client() -> None:
     def _client_factory(**kwargs: Any) -> client_base.BaseClient:
         client: TelnetClient
         kwargs["gmcp_modules"] = gmcp_modules
-        if sys.stdin.isatty():
+        if accessories.is_a_tty():
             client = TelnetTerminalClient(**kwargs)
         else:
             client = TelnetClient(**kwargs)

telnetlib3/tests/test_client_unit.py

@@ -1,4 +1,6 @@
 # std imports
+import io
+import os
 import sys
 import types
 import asyncio
@@ -224,7 +226,7 @@ def test_transform_args_typescript():
 
 @pytest.mark.asyncio
 async def test_open_connection_default_factory(bind_host, unused_tcp_port, monkeypatch):
-    monkeypatch.setattr(sys.stdin, "isatty", lambda: False)
+    monkeypatch.setattr(cl.accessories, "is_a_tty", lambda: False)
 
     async with create_server(host=bind_host, port=unused_tcp_port, connect_maxwait=0.5):
         reader, writer = await cl.open_connection(
@@ -235,10 +237,9 @@ async def test_open_connection_default_factory(bind_host, unused_tcp_port, monke
         writer.close()
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="TTY factory not used on win32")
 @pytest.mark.asyncio
 async def test_open_connection_tty_factory(bind_host, unused_tcp_port, monkeypatch):
-    monkeypatch.setattr(sys.stdin, "isatty", lambda: True)
+    monkeypatch.setattr(cl.accessories, "is_a_tty", lambda: True)
 
     async with create_server(host=bind_host, port=unused_tcp_port, connect_maxwait=0.5):
         reader, writer = await cl.open_connection(
@@ -248,6 +249,76 @@ async def test_open_connection_tty_factory(bind_host, unused_tcp_port, monkeypat
         writer.close()
 
 
+@pytest.mark.asyncio
+async def test_open_connection_without_stdin(bind_host, unused_tcp_port, monkeypatch):
+    """Processes without a console, sys.stdin is None, such as pythonw.exe."""
+    monkeypatch.setattr(sys, "stdin", None)
+
+    async with create_server(host=bind_host, port=unused_tcp_port, connect_maxwait=0.5):
+        reader, writer = await cl.open_connection(
+            host=bind_host, port=unused_tcp_port, connect_maxwait=0.1, encoding=False
+        )
+        assert isinstance(writer.protocol, cl.TelnetClient)
+        assert not isinstance(writer.protocol, cl.TelnetTerminalClient)
+        writer.close()
+
+
+def test_is_a_tty_without_stdin(monkeypatch):
+    monkeypatch.setattr(sys, "stdin", None)
+    assert accessories.is_a_tty() is False
+
+
+def test_is_a_tty_stdin_without_isatty(monkeypatch):
+    monkeypatch.setattr(sys, "stdin", object())
+    assert accessories.is_a_tty() is False
+
+
+def test_is_a_tty_stdin_not_a_tty(monkeypatch):
+    monkeypatch.setattr(sys, "stdin", io.StringIO())
+    assert accessories.is_a_tty() is False
+
+
+def test_is_a_tty_stdin_detached(monkeypatch):
+    class _Detached:
+        def isatty(self):
+            return True
+
+        def fileno(self):
+            raise ValueError("underlying buffer has been detached")
+
+    monkeypatch.setattr(sys, "stdin", _Detached())
+    assert accessories.is_a_tty() is False
+
+
+def test_is_a_tty_stdin_unsupported_fileno(monkeypatch):
+    class _NoFileno:
+        def isatty(self):
+            return True
+
+    monkeypatch.setattr(sys, "stdin", _NoFileno())
+    assert accessories.is_a_tty() is False
+
+
+def test_is_a_tty_stdin_is_a_tty(monkeypatch):
+    class _Tty:
+        def isatty(self):
+            return True
+
+        def fileno(self):
+            # a file descriptor that os.isatty() agrees is a terminal
+            return _Tty.fd
+
+    pty = pytest.importorskip("pty")
+    master_fd, slave_fd = pty.openpty()
+    _Tty.fd = slave_fd
+    try:
+        monkeypatch.setattr(sys, "stdin", _Tty())
+        assert accessories.is_a_tty() is True
+    finally:
+        os.close(slave_fd)
+        os.close(master_fd)
+
+
 def test_detect_syncterm_font_sets_force_binary():
     client = BaseClient.__new__(BaseClient)
     client.log = types.SimpleNamespace(debug=lambda *a, **kw: None, isEnabledFor=lambda _: False)