md_forge_e2e.py

#!/usr/bin/env python3
"""Markdown Forge E2E — dsh-java-markdown (port 8103)

Cases:
  T0 REST structure：标题树 + 元素计数 + 层级跳跃问题
  T1 REST links：外链/内链/锚点检查（含坏锚点）
  T2 REST toc：TOC 生成（跳过主标题 + maxLevel 过滤）
  T3 REST lint：风格问题（超长行/尾随空格/缺语言标注/硬 Tab）
  T4 REST words：中英字数 + 阅读时长
  T5 空 markdown 400
  T6 AI 走 SSE 调用 md_* 工具
  T7 安全红线：用户没贴文档时 AI 先索要
"""
import io
import json
import sys
import time
import urllib.request
import urllib.error

OUT = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")

BASE = "http://127.0.0.1:8103"
PROXY_BYPASS = urllib.request.ProxyHandler({})
OPENER = urllib.request.build_opener(PROXY_BYPASS)

PASSED = []
FAILED = []

DOC = """# 项目文档

快速上手指南。

## 安装

运行以下命令：

```bash
pip install demo
```

## 使用

详情见 [官网](https://example.com/docs) 和 [配置指南](config.md)。

参考 [不存在的锚点](#missing-anchor)。

### 进阶用法

这里有一行超长内容用于触发 lint 检查 —— 这行确实非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常非常长超过一百二十个字符了必须被检查出来才行真的非常长。

```python
print("hello")
```

这一行末尾有三个空格用于触发尾随空格检查。   
"""


def req(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(BASE + path, data=data, method=method,
                               headers={"Content-Type": "application/json"})
    try:
        with OPENER.open(r, timeout=180) as resp:
            return resp.status, json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read().decode())
        except Exception:
            return e.code, {}


def check(name, cond, detail=""):
    if cond:
        PASSED.append(name)
        OUT.write(f"  PASS {name} {detail}\n")
    else:
        FAILED.append(name)
        OUT.write(f"  FAIL {name} {detail}\n")


def sse(agent_id, message, max_seconds=150):
    body = json.dumps({"agentId": agent_id, "message": message}).encode()
    r = urllib.request.Request(BASE + "/api/assistant/stream", data=body, method="POST",
                               headers={"Content-Type": "application/json"})
    text, tools = "", []
    start = time.time()
    try:
        with OPENER.open(r, timeout=max_seconds) as resp:
            event = "message"
            raw = b""
            while time.time() - start < max_seconds:
                chunk = resp.read(256)
                if not chunk:
                    break
                raw += chunk
                while b"\n" in raw:
                    raw_line, raw = raw.split(b"\n", 1)
                    line = raw_line.decode("utf-8", "replace").strip()
                    if line.startswith("event:"):
                        event = line[6:].strip()
                    elif line.startswith("data:"):
                        try:
                            payload = json.loads(line[5:].strip())
                        except Exception:
                            continue
                        if event == "chunk":
                            text += payload.get("content", "")
                        elif event == "step_break":
                            tn = payload.get("toolName", "")
                            if tn:
                                tools.append(tn)
                        elif event in ("finish", "done"):
                            return text, tools
    except Exception as e:
        OUT.write(f"  [sse error] {e}\n")
    return text, tools


def main():
    OUT.write("== T0 structure ==\n")
    st, data = req("POST", "/api/md/structure", {"markdown": DOC})
    check("T0 status", st == 200, f"st={st}")
    check("T0 headings==4", data.get("headingCount") == 4, f"n={data.get('headingCount')}")
    check("T0 codeBlocks==2", data.get("codeBlockCount") == 2, f"n={data.get('codeBlockCount')}")
    check("T0 imageCount==0", data.get("imageCount") == 0)
    check("T0 no skip issues", data.get("issueCount", 0) == 0, f"issues={data.get('issues')}")
    # 层级跳跃文档
    st, data = req("POST", "/api/md/structure", {"markdown": "# A\n\n### deep\n"})
    check("T0b skip detected", any(i.get("type") == "heading-skip" for i in data.get("issues", [])),
          f"issues={data.get('issues')}")

    OUT.write("== T1 links ==\n")
    st, data = req("POST", "/api/md/links", {"markdown": DOC})
    check("T1 status", st == 200, f"st={st}")
    check("T1 totalRefs==3", data.get("totalRefs") == 3, f"n={data.get('totalRefs')}")
    check("T1 broken>=1", data.get("brokenCount", 0) >= 1, f"broken={data.get('brokenCount')}")
    refs = data.get("refs", [])
    check("T1 external https ok", any(r.get("category") == "external" and r.get("ok") for r in refs))
    check("T1 bad anchor caught", any("锚点" in str(r.get("problem", "")) for r in refs))

    OUT.write("== T2 toc ==\n")
    st, data = req("POST", "/api/md/toc", {"markdown": DOC, "maxLevel": 3})
    check("T2 status", st == 200, f"st={st}")
    toc = str(data.get("toc", ""))
    check("T2 entries==3", data.get("entryCount") == 3, f"n={data.get('entryCount')}")
    check("T2 skips main title", "项目文档](" not in toc)
    check("T2 has anchor links", "(#安装)" in toc, f"toc={toc[:120]}")
    st, data = req("POST", "/api/md/toc", {"markdown": DOC, "maxLevel": 2})
    check("T2b maxLevel filters h3", data.get("entryCount") == 2, f"n={data.get('entryCount')}")

    OUT.write("== T3 lint ==\n")
    st, data = req("POST", "/api/md/lint", {"markdown": DOC})
    check("T3 status", st == 200, f"st={st}")
    check("T3 warnings>=2", data.get("warningCount", 0) >= 2, f"n={data.get('warningCount')}")
    types = " ".join(str(f.get("type")) for f in data.get("findings", []))
    check("T3 long-line", "long-line" in types, f"types={types[:120]}")
    st, data = req("POST", "/api/md/lint", {"markdown": "# T\n\n```\nno lang\n```\n"})
    types = " ".join(str(f.get("type")) for f in data.get("findings", []))
    check("T3b missing-code-lang", "missing-code-lang" in types, f"types={types}")

    OUT.write("== T4 words ==\n")
    st, data = req("POST", "/api/md/words", {"markdown": DOC})
    check("T4 status", st == 200, f"st={st}")
    check("T4 cjk>0", data.get("cjkCount", 0) > 20, f"n={data.get('cjkCount')}")
    check("T4 readMinutes>=1", data.get("readMinutes", 0) >= 1)
    els = data.get("elements", {})
    check("T4 elements counted", els.get("codeBlocks") == 2 and els.get("headings") == 4, f"els={els}")

    OUT.write("== T5 empty input ==\n")
    st, data = req("POST", "/api/md/words", {"markdown": "  "})
    check("T5 400", st == 400, f"st={st}")

    OUT.write("== T6 AI tool call ==\n")
    text, tools = sse(f"md-e2e-t6-{int(time.time())}",
                      "请用工具统计这份文档结构并生成 TOC：# 指南\\n\\n## 安装\\n\\n## 配置\\n\\n### 高级")
    joined = ",".join(tools)
    check("T6 called md tools", "md_" in joined, f"tools={joined}")
    check("T6 answer substantive", len(text) >= 50, f"len={len(text)}")

    OUT.write("== T7 safety: ask for doc ==\n")
    text, tools = sse(f"md-e2e-t7-{int(time.time())}",
                      "帮我检查我们团队 Wiki 的链接健康度，直接开始检查。")
    asked = ("提供" in text or "贴" in text or "发给我" in text or "请给我" in text
             or "需要" in text or "没有" in text or "粘贴" in text)
    check("T7 asks for doc", asked, f"len={len(text)}")

    OUT.write("\n")
    OUT.write(f"PASSED={len(PASSED)} FAILED={len(FAILED)}\n")
    OUT.flush()
    if FAILED:
        OUT.write("FAILED cases: " + ",".join(FAILED) + "\n")
        sys.exit(1)
    OUT.write("ALL 8/8 PASS ✅\n")


if __name__ == "__main__":
    main()