FranklinWH-Toolbox/service/agentmcp/agentmcp_test.go

package agentmcp

import (
	"encoding/json"
	"regexp"
	"testing"
	"unicode"

	"fwh-toolbox-server/agentmcp/kit"
)

// 约束测试（agent-mcp.md 3.5）：名称、标题、描述、读写声明、schema 类型。
var nameRE = regexp.MustCompile(`^[a-z][a-z0-9_]{2,39}$`)

// 引擎侧完整名称 mcp__gateway__<服务名>__<工具名> 不能超过 64；服务名最长按 6 个字符（fwhops）计算
const enginePrefixLen = len("mcp__gateway__") + len("fwhops") + len("__")

func TestToolConstraints(t *testing.T) {
	seen := map[string]string{}
	for _, set := range []string{SetReadOnly, SetOps} {
		list, err := Tools(set)
		if err != nil {
			t.Fatal(err)
		}
		if len(list) == 0 {
			t.Fatalf("工具集 %s 为空", set)
		}
		for _, r := range list {
			m := r.Meta()
			if !nameRE.MatchString(m.Name) {
				t.Errorf("%s: 名称不符合 ^[a-z][a-z0-9_]{2,39}$", m.Name)
			}
			if enginePrefixLen+len(m.Name) > 64 {
				t.Errorf("%s: 引擎侧完整名称超过 64 字符", m.Name)
			}
			if prev, dup := seen[m.Name]; dup {
				t.Errorf("%s: 与工具集 %s 中的工具重名", m.Name, prev)
			}
			seen[m.Name] = set
			if m.Title == "" {
				t.Errorf("%s: 缺少 title", m.Name)
			}
			if !hasHan(m.Description) {
				t.Errorf("%s: 描述应为中文", m.Name)
			}
			if set == SetReadOnly && !m.ReadOnly {
				t.Errorf("%s: readonly 工具集中出现写工具", m.Name)
			}
			if set == SetOps && m.ReadOnly {
				t.Errorf("%s: ops 工具集中出现只读工具", m.Name)
			}

			tool := r.Tool()
			if tool.Annotations.ReadOnlyHint == nil {
				t.Errorf("%s: 未显式声明 readOnlyHint", m.Name)
			} else if *tool.Annotations.ReadOnlyHint != m.ReadOnly {
				t.Errorf("%s: readOnlyHint 与定义不一致", m.Name)
			}
			if tool.Annotations.Title != m.Title {
				t.Errorf("%s: title 注解与定义不一致", m.Name)
			}

			b, err := json.Marshal(tool)
			if err != nil {
				t.Fatalf("%s: 序列化失败: %v", m.Name, err)
			}
			var decoded struct {
				InputSchema map[string]any `json:"inputSchema"`
			}
			if err := json.Unmarshal(b, &decoded); err != nil {
				t.Fatalf("%s: 解析失败: %v", m.Name, err)
			}
			if decoded.InputSchema["type"] != "object" {
				t.Errorf("%s: inputSchema.type 应为 object，实际 %v", m.Name, decoded.InputSchema["type"])
			}
		}
	}
}

// 一期清单（agent-mcp.md 第 4 节）全部注册
func TestReadOnlyToolList(t *testing.T) {
	want := []string{
		"site_search", "site_get", "site_power_flow", "site_power_flow_export",
		"alarm_current", "alarm_events", "alarm_snapshot", "alarm_stats", "alarm_sites_by_code",
		"shutdown_stats", "shutdown_sites",
		"mqtt_log_search", "ems_log_records", "update_log_system", "update_log_ahub",
		"fault_record_list", "fault_record_detail", "fault_record_callable", "fault_record_call_history",
		"tou_template_list", "tou_template_get",
		"sim_get", "sim_search", "sim_usage_monthly",
		"ems_param_lookup", "plane_issue_search", "plane_issue_get", "sn_parse",
	}
	list, _ := Tools(SetReadOnly)
	got := map[string]bool{}
	for _, r := range list {
		got[r.Meta().Name] = true
	}
	for _, n := range want {
		if !got[n] {
			t.Errorf("缺少工具 %s", n)
		}
	}
	if len(got) != len(want) {
		t.Errorf("只读工具数量 %d，期望 %d", len(got), len(want))
	}
}

func TestExportTimeoutByName(t *testing.T) {
	if !(kit.Meta{Name: "site_power_flow_export"}).IsExport() {
		t.Error("site_power_flow_export 应识别为导出工具")
	}
	if (kit.Meta{Name: "site_power_flow"}).IsExport() {
		t.Error("site_power_flow 不是导出工具")
	}
}

func hasHan(s string) bool {
	for _, r := range s {
		if unicode.Is(unicode.Han, r) {
			return true
		}
	}
	return false
}