diff --git a/src/keboola_agent_cli/commands/serve.py b/src/keboola_agent_cli/commands/serve.py index b890d2eb..3a2090be 100644 --- a/src/keboola_agent_cli/commands/serve.py +++ b/src/keboola_agent_cli/commands/serve.py @@ -86,6 +86,53 @@ def _autodetect_ui_dist() -> Path | None: return None +# Box-drawing glyphs used by the startup banner, mapped to ASCII so the banner +# degrades to `|-` / `` `- `` on a console that can't render Unicode -- the same +# fallback install.sh uses (Unicode only under a UTF-8 locale, ASCII otherwise). +_BANNER_ASCII_FALLBACK = str.maketrans({"├": "|", "└": "`", "─": "-"}) + + +def _encode_safe(text: str, encoding: str | None) -> str: + """Return ``text`` with box-drawing glyphs stripped to ASCII if unencodable. + + On Windows with a non-UTF-8 console codepage (cp1250 on Czech/Polish/ + Hungarian locales, and any other legacy single-byte encoding) the ``├─`` / + ``└─`` glyphs in the startup banner cannot be encoded, so ``sys.stdout.write`` + raises ``UnicodeEncodeError`` -- which crashed ``kbagent serve`` before + uvicorn ever bound the port (issue #522). When ``encoding`` can represent the + banner we keep the Unicode glyphs; otherwise we transliterate to ASCII. + + ``encoding`` is ``sys.stdout.encoding`` at the call site (``None`` on an + exotic stream); an unknown codec name (``LookupError``) also degrades to + ASCII rather than propagating. + """ + enc = encoding or "utf-8" + try: + text.encode(enc) + except (UnicodeEncodeError, LookupError): + return text.translate(_BANNER_ASCII_FALLBACK) + return text + + +def _write_banner(text: str) -> None: + """Write the startup banner without ever letting a glyph crash startup. + + Two layers of defense (issue #522): first transliterate to ASCII when + stdout's encoding can't represent the banner, then wrap the write in a + belt-and-braces ``try/except`` so even an unforeseen un-encodable character + re-emits with lossy replacement instead of aborting the server. A degraded + banner is always better than a server that won't start. + """ + encoding = getattr(sys.stdout, "encoding", None) + safe = _encode_safe(text, encoding) + try: + sys.stdout.write(safe) + except UnicodeEncodeError: + enc = encoding or "ascii" + sys.stdout.write(safe.encode(enc, "replace").decode(enc, "replace")) + sys.stdout.flush() + + def serve_command( host: str = typer.Option( "127.0.0.1", @@ -232,7 +279,7 @@ def serve_command( # UI mode: the user opens the browser directly; there is no BFF to # paste the token into. The token is still printed so scripted # callers / curl one-liners keep working. - sys.stdout.write( + banner = ( "\n" " kbagent serve (single-process UI mode)\n" f" ├─ open: http://{host}:{port}/\n" @@ -252,7 +299,7 @@ def serve_command( "\n" ) else: - sys.stdout.write( + banner = ( "\n" " kbagent serve\n" f" ├─ host: http://{host}:{port}\n" @@ -270,7 +317,7 @@ def serve_command( f" export KBAGENT_CONVERSATION_ID={conversation_id}\n" "\n" ) - sys.stdout.flush() + _write_banner(banner) uvicorn.run( app, diff --git a/tests/test_serve_banner_encoding.py b/tests/test_serve_banner_encoding.py new file mode 100644 index 00000000..17c0e6c2 --- /dev/null +++ b/tests/test_serve_banner_encoding.py @@ -0,0 +1,252 @@ +"""Tests for the encoding-safe startup banner in ``kbagent serve`` (issue #522). + +On Windows with a non-UTF-8 console codepage (cp1250 on Czech/Polish/Hungarian +locales, and any legacy single-byte encoding) the box-drawing glyphs ``├─`` / +``└─`` in the startup banner cannot be encoded. ``sys.stdout.write`` then raises +``UnicodeEncodeError`` -- and because that happens *before* ``uvicorn.run``, the +server never binds the port. The banner is cosmetic, so it must degrade to ASCII +instead of crashing startup. + +These tests cover the two helpers that implement the fix: + +* ``_encode_safe(text, encoding)`` -- pure transliteration decision. +* ``_write_banner(text)`` -- the I/O wrapper with a belt-and-braces fallback. + +plus an AST-based regression guard that keeps the ASCII fallback table in sync +with whatever glyphs the live banner literals actually use. +""" + +from __future__ import annotations + +import ast +import sys +from pathlib import Path + +import pytest + +from keboola_agent_cli.commands import serve +from keboola_agent_cli.commands.serve import ( + _BANNER_ASCII_FALLBACK, + _encode_safe, + _write_banner, +) + +# The exact glyphs the banner uses; cp1250 (and every other legacy single-byte +# console codepage) cannot encode any of them. +_BOX_GLYPHS = ("├", "└", "─") + + +class _Cp1250Stdout: + """A stdout stand-in that behaves like a real cp1250 Windows console. + + Writing a character cp1250 cannot represent raises ``UnicodeEncodeError``, + exactly as the Windows console does -- so a test that drives ``_write_banner`` + through this object reproduces issue #522 faithfully rather than approximating + it. + """ + + encoding = "cp1250" + + def __init__(self) -> None: + self._chunks: list[str] = [] + + def write(self, s: str) -> int: + s.encode(self.encoding) # raises UnicodeEncodeError on ├ └ ─ + self._chunks.append(s) + return len(s) + + def flush(self) -> None: # pragma: no cover - trivial + pass + + def getvalue(self) -> str: + return "".join(self._chunks) + + +class TestFakeConsoleFidelity: + """Guard the guard: prove the fake actually reproduces the crash, so the + passing ``_write_banner`` test below is meaningful and not vacuous.""" + + def test_naive_write_of_glyphs_crashes(self) -> None: + fake = _Cp1250Stdout() + with pytest.raises(UnicodeEncodeError): + fake.write(" ├─ open: http://x\n") + + def test_plain_ascii_write_is_fine(self) -> None: + fake = _Cp1250Stdout() + fake.write(" |- open: http://x\n") + assert "open" in fake.getvalue() + + +class TestEncodeSafe: + def test_utf8_keeps_unicode_glyphs(self) -> None: + banner = " ├─ open\n └─ token\n" + # A UTF-8 console can render the glyphs, so nothing is transliterated. + assert _encode_safe(banner, "utf-8") == banner + + def test_cp1250_transliterates_to_ascii(self) -> None: + out = _encode_safe(" ├─ open\n └─ token\n", "cp1250") + assert out == " |- open\n `- token\n" + for glyph in _BOX_GLYPHS: + assert glyph not in out + + def test_none_encoding_defaults_to_utf8(self) -> None: + # An exotic stream can report ``encoding is None``; we default to UTF-8 + # (the modern norm) and keep the glyphs rather than needlessly degrading. + banner = " ├─ open\n" + assert _encode_safe(banner, None) == banner + + def test_unknown_codec_degrades_to_ascii(self) -> None: + # A bogus/unknown codec name raises LookupError from str.encode; that + # must degrade to ASCII, not propagate. + out = _encode_safe(" ├─ open\n", "definitely-not-a-codec") + assert out == " |- open\n" + + def test_ascii_only_text_is_unchanged_on_cp1250(self) -> None: + text = " kbagent serve\n token: abc123\n" + assert _encode_safe(text, "cp1250") == text + + @pytest.mark.parametrize("encoding", ["cp1250", "cp1252", "latin-1", "ascii"]) + def test_result_is_always_encodable_on_legacy_consoles(self, encoding: str) -> None: + # Property: whatever legacy single-byte console we target, the result + # of _encode_safe must round-trip cleanly (no UnicodeEncodeError). + banner = " ├─ open\n ├─ docs\n └─ token\n" + _encode_safe(banner, encoding).encode(encoding) # must not raise + + +class TestWriteBanner: + def test_does_not_crash_on_cp1250_and_emits_ascii( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + fake = _Cp1250Stdout() + monkeypatch.setattr(sys, "stdout", fake) + _write_banner(" ├─ open: http://127.0.0.1:8001/\n └─ token: secret\n") + out = fake.getvalue() + assert "|- open" in out + assert "`- token" in out + for glyph in _BOX_GLYPHS: + assert glyph not in out + + def test_preserves_unicode_on_utf8_console(self, monkeypatch: pytest.MonkeyPatch) -> None: + # A real UTF-8 console keeps the glyphs verbatim -- the fix must not + # degrade output on the common (modern) case. (io.StringIO can't stand + # in here: it's a slotted C type with no settable ``encoding`` attr.) + class _Utf8Stdout: + encoding = "utf-8" + + def __init__(self) -> None: + self._chunks: list[str] = [] + + def write(self, s: str) -> int: + s.encode(self.encoding) # utf-8 encodes everything -> never raises + self._chunks.append(s) + return len(s) + + def flush(self) -> None: + pass + + def getvalue(self) -> str: + return "".join(self._chunks) + + buf = _Utf8Stdout() + monkeypatch.setattr(sys, "stdout", buf) + _write_banner(" ├─ open\n └─ token\n") + assert buf.getvalue() == " ├─ open\n └─ token\n" + + def test_belt_and_braces_survives_liar_encoding(self, monkeypatch: pytest.MonkeyPatch) -> None: + # Pathological stream: claims UTF-8 (so _encode_safe keeps the glyphs) + # but its write() still rejects them. The try/except must catch the + # UnicodeEncodeError and re-emit lossily rather than crash startup. + class _LiarStdout: + encoding = "utf-8" + + def __init__(self) -> None: + self.calls = 0 + self.buf: list[str] = [] + + def write(self, s: str) -> int: + self.calls += 1 + if self.calls == 1: + # First (optimistic) write rejects the glyphs. + raise UnicodeEncodeError("cp1250", s, 0, 1, "mock") + self.buf.append(s) + return len(s) + + def flush(self) -> None: + pass + + liar = _LiarStdout() + monkeypatch.setattr(sys, "stdout", liar) + _write_banner(" ├─ open\n") # must not raise + assert liar.calls == 2 # optimistic write, then lossy fallback write + + +def _static_banner_text(node: ast.AST) -> str: + """Flatten a str/f-string literal AST node to its static text. + + ``banner = ("...\n" f" ├─ {x}\n" ...)`` parses to a ``Constant`` (pure + string) or ``JoinedStr`` (f-string). We keep the literal ``Constant`` parts + (which carry the box glyphs) and drop the ``{expr}`` interpolations, which + are runtime values with no bearing on the fallback table. + """ + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.JoinedStr): + parts = [] + for value in node.values: + if isinstance(value, ast.Constant) and isinstance(value.value, str): + parts.append(value.value) + # FormattedValue ({expr}) contributes only runtime data -> skip. + return "".join(parts) + return "" + + +def _live_banner_literals() -> list[str]: + """Extract every ``banner = (...)`` literal from serve_command's source. + + Parsing the real module (not a copy pasted into the test) means this guard + tracks whatever glyphs the shipping banner actually uses -- add a new one + without extending _BANNER_ASCII_FALLBACK and the round-trip below fails. + """ + src = Path(serve.__file__).read_text(encoding="utf-8") + tree = ast.parse(src) + fn = next( + n for n in ast.walk(tree) if isinstance(n, ast.FunctionDef) and n.name == "serve_command" + ) + literals = [ + _static_banner_text(node.value) + for node in ast.walk(fn) + if isinstance(node, ast.Assign) + and any(isinstance(t, ast.Name) and t.id == "banner" for t in node.targets) + ] + return [text for text in literals if text] + + +class TestLiveBannerRegression: + """The fallback table must cover every glyph the real banner emits.""" + + def test_both_banners_were_found(self) -> None: + # Sanity: serve_command builds two banners (UI and API-only). If a + # refactor renames the variable, catch it here rather than silently + # testing nothing. + literals = _live_banner_literals() + assert len(literals) == 2, f"expected 2 banner literals, found {len(literals)}" + + @pytest.mark.parametrize("encoding", ["cp1250", "cp1252", "latin-1"]) + def test_live_banners_survive_legacy_consoles(self, encoding: str) -> None: + for banner in _live_banner_literals(): + safe = _encode_safe(banner, encoding) + safe.encode(encoding) # must not raise -> banner can't crash serve + + def test_every_non_ascii_glyph_is_in_fallback_table(self) -> None: + # Directly assert the invariant: any non-ASCII character appearing in a + # live banner literal has an ASCII mapping. This is the actionable + # failure ("add glyph X to _BANNER_ASCII_FALLBACK") for a future editor. + # str.maketrans keys the table by code point (int), not by character, + # so membership is tested with ord(ch). + for banner in _live_banner_literals(): + for ch in banner: + if ord(ch) > 127: + assert ord(ch) in _BANNER_ASCII_FALLBACK, ( + f"banner glyph {ch!r} (U+{ord(ch):04X}) has no ASCII " + f"fallback; add it to _BANNER_ASCII_FALLBACK in serve.py" + )