audio_codebuddy/mcp-for-xiaozhi/tests/chapter-read.test.py

"""
MCP read_chapter 工具测试 + tools/list schema 字节断言

覆盖：
  1. schema 总字节 < 1024（小智限制）
  2. read_chapter 客户端参数校验
  3. format=text → RESPONSE 模式
  4. format=audio（已有 full 音频）→ 返回 audio_url
  5. format=audio（on_demand）→ 返回 audio_source=on_demand
  6. 后端 code=2002 → MCP 自动降级到 text
  7. 后端 code=2001 → _err
  8. TOOL_DISPATCH / TOOLS 注册正确
"""
from __future__ import annotations
import json
import sys
import os

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.dirname(HERE))

import audio_server as A


def parse_result(r):
    """解 _ok 包装：{success, result: '<json-string>'} → dict"""
    if r.get("success") and r.get("result"):
        try:
            return json.loads(r["result"])
        except Exception:
            return {}
    return {}


def fake_http_get(path, params=None):
    return {
        "code": fake_http_get.code,
        "message": fake_http_get.msg or "success",
        "data": fake_http_get.data,
    }


A._http_get = fake_http_get


def reset(code=0, data=None, msg=None):
    fake_http_get.code = code
    fake_http_get.data = data
    fake_http_get.msg = msg


def assert_true(cond, msg):
    if not cond:
        raise AssertionError("FAIL: " + msg)
    print("[PASS]", msg)


# ============ 测试用例 ============

def test_schema_size():
    size = len(json.dumps(A.TOOLS, ensure_ascii=False).encode("utf-8"))
    assert_true(size < 1024, f"tools/list < 1024 bytes (actual {size})")


def test_required_chapter_id():
    r = A.tool_read_chapter({})
    assert_true(r.get("success") is False, "no chapter_id -> _err")


def test_invalid_format():
    r = A.tool_read_chapter({"chapter_id": 1, "format": "xml"})
    assert_true(r.get("success") is False, "invalid format -> _err")


def test_format_text():
    reset(code=0, data={
        "mode": "text", "chapter_id": 100, "book_id": 1, "title": "DiYiHui",
        "text": "hua shuo tian xia da shi ...", "has_more": False, "next_offset": None,
        "nav": {"chapter_index": 1, "total_chapters": 5, "prev_chapter_id": None, "next_chapter_id": 101},
    })
    r = A.tool_read_chapter({"chapter_id": 100, "format": "text"})
    assert_true(r.get("action") == "RESPONSE", "format=text -> action=RESPONSE")
    assert_true("DiYiHui" in r["response"], "response contains title")
    assert_true(r.get("next_chapter_id") == 101, "nav next_chapter_id present")


def test_format_audio_full():
    reset(code=0, data={
        "mode": "audio", "chapter_id": 100, "book_id": 1, "title": "T",
        "audio_url": "https://oss.example.com/full.mp3",
        "audio_duration": 320, "audio_source": "full",
        "nav": {"chapter_index": 1, "total_chapters": 5, "prev_chapter_id": None, "next_chapter_id": 101},
    })
    r = A.tool_read_chapter({"chapter_id": 100, "format": "audio"})
    assert_true(r.get("success") is True, "format=audio -> success")
    inner = parse_result(r)
    assert_true(inner.get("audio_url") == "https://oss.example.com/full.mp3", "audio_url correct")
    assert_true(inner.get("audio_source") == "full", "audio_source=full")
    assert_true("read_chapter" in inner.get("action", ""), "nav action hint")


def test_format_audio_on_demand():
    reset(code=0, data={
        "mode": "audio", "chapter_id": 100, "book_id": 1, "title": "T",
        "audio_url": "https://oss.example.com/sync.mp3",
        "audio_duration": 30, "audio_source": "on_demand",
        "nav": {"chapter_index": 1, "total_chapters": 1, "prev_chapter_id": None, "next_chapter_id": None},
    })
    r = A.tool_read_chapter({"chapter_id": 100, "format": "audio"})
    inner = parse_result(r)
    assert_true(inner.get("audio_source") == "on_demand", "audio_source=on_demand")
    assert_true(inner.get("action", "") != "", "non-empty action for last chapter")


def test_tts_fallback_to_text():
    """后端 code=2002 -> MCP 自动降级到 format=text"""
    call_count = [0]

    def mock_with_fallback(path, params=None):
        call_count[0] += 1
        if call_count[0] == 1:
            return {"code": 2002, "message": "TTS_FALLBACK_FAILED", "data": None}
        return {"code": 0, "message": "success", "data": {
            "mode": "text", "chapter_id": 100, "book_id": 1, "title": "T",
            "text": "fallback text", "has_more": False, "next_offset": None,
            "nav": {"chapter_index": 1, "total_chapters": 1, "prev_chapter_id": None, "next_chapter_id": None},
        }}

    A._http_get = mock_with_fallback
    r = A.tool_read_chapter({"chapter_id": 100, "format": "audio"})
    assert_true(r.get("action") == "RESPONSE", "2002 fallback -> RESPONSE mode")
    assert_true("fallback" in r.get("response", "").lower(), "fallback text content")
    assert_true(call_count[0] == 2, "called 2 times (retry after failure)")
    A._http_get = fake_http_get


def test_other_error_returns_err():
    A._http_get = lambda p, q=None: {"code": 2001, "message": "CHAPTER_NOT_FOUND", "data": None}
    r = A.tool_read_chapter({"chapter_id": 999, "format": "audio"})
    assert_true(r.get("success") is False, "2001 -> _err")
    A._http_get = fake_http_get


def test_dispatch_registered():
    assert_true("read_chapter" in A.TOOL_DISPATCH, "TOOL_DISPATCH has read_chapter")
    assert_true(any(t["name"] == "read_chapter" for t in A.TOOLS), "TOOLS has read_chapter")


def main():
    tests = [
        test_schema_size, test_required_chapter_id, test_invalid_format,
        test_format_text, test_format_audio_full, test_format_audio_on_demand,
        test_tts_fallback_to_text, test_other_error_returns_err, test_dispatch_registered,
    ]
    passed = 0
    failed = 0
    for t in tests:
        try:
            t()
            passed += 1
        except AssertionError as e:
            failed += 1
            print("[FAIL]", t.__name__, "->", e)
    print(f"\n{passed} passed, {failed} failed")
    sys.exit(0 if failed == 0 else 1)


if __name__ == "__main__":
    main()