diff --git a/docs/history.rst b/docs/history.rst index be8835f..658938a 100644 --- a/docs/history.rst +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 5d4ed9a..e7a3f51 100644 --- a/pyproject.toml +++ b/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" diff --git a/telnetlib3/accessories.py b/telnetlib3/accessories.py index 34ccb03..b1c0cf3 100644 --- a/telnetlib3/accessories.py +++ b/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 diff --git a/telnetlib3/client.py b/telnetlib3/client.py index 08add24..7ed118a 100755 --- a/telnetlib3/client.py +++ b/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) diff --git a/telnetlib3/tests/test_client_unit.py b/telnetlib3/tests/test_client_unit.py index 0393587..8f56d2a 100644 --- a/telnetlib3/tests/test_client_unit.py +++ b/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)