visionA/local-tool/server/scripts/test_kneron_bridge_encoding.py

#!/usr/bin/env python3
"""Unit tests for kneron_bridge stdio encoding (Windows 中文亂碼修正).

背景 —— 為什麼需要這組測試：

Windows 的 Python 預設把 stdin / stdout / stderr 綁到「系統 ANSI code page」
而非 UTF-8。繁體中文 Windows 的 ANSI code page 是 cp950。JSON-RPC 兩個方向
的行為並不對稱：

  Go → Python (stdin)：Go 的 `encoding/json` **不** escape 非 ASCII，
      `{"labels":["布"]}` 在 wire 上是 raw UTF-8 位元組 e5 b8 83。
      Python 端若以 cp950 解碼會得到 `撣` + U+FFFD —— 這正是使用者截圖
      看到的亂碼。嚴格模式下則直接 UnicodeDecodeError 讓 bridge 整個掛掉。

  Python → Go (stdout)：`json.dumps` 預設 ensure_ascii=True，中文被 escape
      成 \\uXXXX 純 ASCII，所以 JSON 回應這條路「目前」不會壞。但這是脆弱的
      隱性依賴：任何人加上 ensure_ascii=False 就會壞。stderr 上的 `_log()`
      中文訊息則是現在就會壞（使用者先前看到的 `<60>X SDK did not report`
      就是 em dash U+2014 被非 UTF-8 編碼破壞的結果）。

因此修正必須同時涵蓋 stdin（decode）與 stdout / stderr（encode）。

測試策略 —— 如何在 macOS 上測 Windows-only 的 bug：
本機 locale 是 UTF-8，直接跑不會重現。改用 `PYTHONIOENCODING` 環境變數強制
子行程的 stdio 編碼，模擬繁中 Windows 的預設行為。修正正確時，bridge 必須
忽略這個「錯誤的」預設值、一律以 UTF-8 處理 stdio。

為什麼用 `unknown command` 這條路徑做斷言：
它是唯一「不需要實體 Kneron dongle、又會把輸入字串原樣 echo 回 stdout」的
handler，因此能驗證完整的 stdin → 解析 → stdout 往返。真正的 labels 路徑
（set_inference_options）需要已連線的裝置，在 CI / 本機無法觸及；但兩者共用
同一個 stdin 解碼器，所以這裡測到的就是同一個根因。

執行方式：
    cd server/scripts && python3 -m unittest test_kneron_bridge_encoding
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
BRIDGE = os.path.join(HERE, "kneron_bridge.py")

# 取自使用者實際的 labels.txt。`布` 特別重要：UTF-8 位元組 (e5 b8 83) 以
# cp950 解碼會得到 `撣` + U+FFFD，正是截圖中的症狀。
CJK_LABELS = ["剪刀", "石頭", "布"]

# U+FFFD REPLACEMENT CHARACTER —— 解碼失敗的明確指紋。
REPLACEMENT_CHAR = "<EFBFBD>"

# 模擬非 UTF-8 系統預設編碼的情境。
#   cp950          —— 繁中 Windows 的 ANSI code page（嚴格模式會拋例外）
#   cp950:replace  —— 靜默產生 U+FFFD 的最陰險情境（使用者實際看到的）
#   latin-1        —— 確認修正不是只針對 cp950 硬編
NON_UTF8_STDIO_ENVS = ["cp950", "cp950:replace", "latin-1"]


def _run_bridge(commands, io_encoding=None, timeout=60):
    """把 commands 逐行餵給 bridge 子行程，回傳 (responses, proc)。

    刻意走真實 subprocess 而非 import：這個 bug 的本質就是「行程啟動時
    stdio 綁定了什麼編碼」，in-process 測試無法重現。

    以 bytes 模式通訊：測試自己控制編解碼，才能精準模擬 Go 端 json.Marshal
    送出 raw UTF-8 的行為，不受測試進程自身 locale 影響。
    """
    env = os.environ.copy()
    # 避免使用者環境既有的設定干擾測試前提。
    env.pop("PYTHONIOENCODING", None)
    if io_encoding is not None:
        env["PYTHONIOENCODING"] = io_encoding

    payload = b"".join(
        json.dumps(c, ensure_ascii=False).encode("utf-8") + b"\n" for c in commands
    )

    proc = subprocess.run(
        [sys.executable, BRIDGE],
        input=payload,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        env=env,
        timeout=timeout,
    )

    responses = []
    for raw in proc.stdout.splitlines():
        raw = raw.strip()
        if not raw:
            continue
        try:
            responses.append(json.loads(raw.decode("utf-8")))
        except (UnicodeDecodeError, json.JSONDecodeError):
            # 保留原始 bytes，讓失敗訊息看得出是編碼問題而非邏輯問題。
            responses.append({"_undecodable": raw})
    return responses, proc


class StdinDecodingTest(unittest.TestCase):
    """stdin 方向：Go 送 raw UTF-8，bridge 必須正確解碼。"""

    def _assert_cjk_roundtrip(self, io_encoding, scenario):
        # 每個標籤各送一次，確保是逐字驗證而非整包碰運氣。
        commands = [{"cmd": f"probe_{label}"} for label in CJK_LABELS]
        responses, proc = _run_bridge(commands, io_encoding)

        stderr_text = proc.stderr.decode("utf-8", "replace")

        self.assertEqual(
            proc.returncode, 0,
            msg=(f"[{scenario}] bridge 非正常結束（rc={proc.returncode}）。"
                 f"多半是 stdin 以非 UTF-8 解碼導致 UnicodeDecodeError。\n"
                 f"stderr=\n{stderr_text}"),
        )

        for r in responses:
            self.assertNotIn(
                "_undecodable", r,
                msg=f"[{scenario}] stdout 不是合法 UTF-8 JSON：{r}",
            )

        errors = [r["error"] for r in responses if isinstance(r.get("error"), str)]
        self.assertEqual(
            len(errors), len(CJK_LABELS),
            msg=(f"[{scenario}] 預期 {len(CJK_LABELS)} 筆 unknown-command 回應，"
                 f"實得 {len(errors)}。responses={responses}\n"
                 f"stderr=\n{stderr_text}"),
        )

        for label, err in zip(CJK_LABELS, errors):
            self.assertNotIn(
                REPLACEMENT_CHAR, err,
                msg=(f"[{scenario}] 回應含 U+FFFD replacement character，"
                     f"代表 stdin 被以非 UTF-8 解碼：{err!r}"),
            )
            self.assertIn(
                label, err,
                msg=(f"[{scenario}] 中文字串 round-trip 失敗。"
                     f"預期回應含 {label!r}，實得 {err!r}"),
            )

    def test_roundtrip_under_default_locale(self):
        """基準線：本機預設 locale（macOS / Linux 為 UTF-8）必須正常。

        這條在修正前後都會過 —— 它的作用是證明測試本身沒問題，
        失敗的是編碼而不是斷言寫錯。
        """
        self._assert_cjk_roundtrip(None, "default-locale")

    def test_roundtrip_under_non_utf8_stdio(self):
        """核心回歸測試：模擬繁中 Windows 的非 UTF-8 預設 stdio。

        修正前：cp950 → UnicodeDecodeError 直接 crash；
                cp950:replace → 靜默產生 `撣<>`。
        修正後：bridge 強制 UTF-8，忽略 PYTHONIOENCODING 的錯誤指示。
        """
        for io_encoding in NON_UTF8_STDIO_ENVS:
            with self.subTest(encoding=io_encoding):
                self._assert_cjk_roundtrip(io_encoding, io_encoding)


class StderrEncodingTest(unittest.TestCase):
    """stderr 方向：_log() 的中文 / 非 ASCII 診斷訊息必須可讀。"""

    def test_stderr_is_valid_utf8_under_non_utf8_stdio(self):
        """使用者先前看到的 `<60>X SDK did not report` 與 label 亂碼同源。

        stderr 若被綁到 cp950，em dash（U+2014）等非 ASCII 字元會被破壞，
        讓診斷訊息失去價值。
        """
        for io_encoding in NON_UTF8_STDIO_ENVS:
            with self.subTest(encoding=io_encoding):
                _, proc = _run_bridge([{"cmd": "probe_布"}], io_encoding)
                try:
                    proc.stderr.decode("utf-8")
                except UnicodeDecodeError as e:
                    self.fail(
                        f"[{io_encoding}] stderr 不是合法 UTF-8，"
                        f"編碼修正未涵蓋 stderr：{e}")

    def test_startup_banner_present(self):
        """確認 bridge 真的有跑起來（避免上面的測試因空輸出而假綠）。"""
        _, proc = _run_bridge([{"cmd": "probe_布"}], "cp950:replace")
        self.assertIn(
            "[kneron_bridge]", proc.stderr.decode("utf-8", "replace"),
            msg="bridge 啟動 log 不存在，測試前提不成立")


class StdoutEncodingTest(unittest.TestCase):
    """stdout 方向：JSON 回應必須能以 UTF-8 解碼。"""

    def test_stdout_is_valid_utf8_with_non_ascii_payload(self):
        """即使未來有人把 json.dumps 改成 ensure_ascii=False 也不能壞。

        目前 ensure_ascii=True 讓中文變成 \\uXXXX 純 ASCII，掩蓋了 stdout
        的編碼問題。這條測試把「stdout 必須是 UTF-8」釘成顯式契約，
        而不是依賴 json.dumps 的預設值。
        """
        for io_encoding in NON_UTF8_STDIO_ENVS:
            with self.subTest(encoding=io_encoding):
                responses, proc = _run_bridge(
                    [{"cmd": "probe_布"}], io_encoding)
                for r in responses:
                    self.assertNotIn(
                        "_undecodable", r,
                        msg=(f"[{io_encoding}] stdout 無法以 UTF-8 解碼：{r}"))


class RealStdoutFdEncodingTest(unittest.TestCase):
    """main() 裡 os.fdopen() 建立的 JSON-RPC 回應通道必須顯式綁 UTF-8。

    ⚠️ 這條為什麼是原始碼層級的斷言，而不是行為測試：

    `os.fdopen(fd, "w")` 不帶 encoding 時，編碼取自
    `locale.getencoding()`（PEP 686 之後 io.text_encoding(None) → "locale"），
    **不是** `PYTHONIOENCODING`。PYTHONIOENCODING 只影響 sys.std*。

    後果：在 macOS / Linux（locale 為 UTF-8）上，無論怎麼設
    PYTHONIOENCODING 都無法讓這個檔案物件變成 cp950，因此
    上面的 subprocess 測試**結構上碰不到這條路徑**（已用 mutation test
    證實：拿掉 encoding= 參數後那些測試仍然全綠）。

    但在繁中 Windows 上 locale.getencoding() 就是 cp950，這個 fd
    會真的以 cp950 編碼寫出 JSON 回應。目前 json.dumps 的
    ensure_ascii=True 讓中文變成純 ASCII 而僥倖沒出事 —— 這是巧合，
    不是設計。

    唯一能在 macOS 上守住這條的方式，就是把「必須顯式指定 encoding」
    當成不可回歸的原始碼契約來檢查。
    """

    def test_fdopen_specifies_utf8_explicitly(self):
        with open(BRIDGE, encoding="utf-8") as f:
            source = f.read()

        self.assertIn(
            "_real_stdout_fd = os.dup(1)", source,
            msg="main() 的 stdout dup 結構已改變，本測試需同步更新")

        # 找出建立 _real_stdout 的那個 os.fdopen 呼叫（可能跨行）。
        marker = "_real_stdout = os.fdopen("
        idx = source.find(marker)
        self.assertNotEqual(
            idx, -1, msg="找不到 _real_stdout = os.fdopen(...) 呼叫")

        # 取到該敘述結束（右括號）為止，避免誤抓到後面無關的程式碼。
        call = source[idx:source.find(")", idx) + 1]

        self.assertIn(
            'encoding="utf-8"', call,
            msg=(f"os.fdopen() 未顯式指定 encoding=\"utf-8\"，"
                 f"在繁中 Windows 上會退回 cp950 編碼 JSON-RPC 回應。"
                 f"實際呼叫：{call!r}"))


if __name__ == "__main__":
    unittest.main(verbosity=2)