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", "site_info_sync_status", "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", "mqtt_traffic_trend", "mqtt_traffic_sites", "mqtt_traffic_site_cmd", "mqtt_message_search", "auto_update_catalog", "auto_update_config_get", "site_analysis_list", "site_analysis_timeline", "site_analysis_export", "fault_record_devices", "fault_record_device_info", "fault_record_param_read", "fault_record_operations", "fault_record_operation_get", "fault_record_data", "fault_record_export", "cloud_fault_record_list", "cloud_fault_record_detail", "cloud_fault_record_callable", "cloud_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 }