feat: add DingTalk directory-backed session titles
This commit is contained in:
姚章浩
@@ -0,0 +1,85 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/directory"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/store/postgres"
|
||||
)
|
||||
|
||||
func main() {
|
||||
if err := run(os.Args[1:], os.Stdout); err != nil {
|
||||
fmt.Fprintln(os.Stderr, err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
func run(args []string, out io.Writer) error {
|
||||
flags := flag.NewFlagSet("directory-import", flag.ContinueOnError)
|
||||
flags.SetOutput(io.Discard)
|
||||
csvPath := flags.String("csv", "", "path to tested DingTalk directory CSV output")
|
||||
tenant := flags.String("tenant", "", "tenant key")
|
||||
databaseURLFile := flags.String("database-url-file", "", "file containing DATABASE_URL")
|
||||
validateOnly := flags.Bool("validate-only", false, "validate CSV without connecting to PostgreSQL")
|
||||
if err := flags.Parse(args); err != nil {
|
||||
return errors.New("invalid directory import arguments")
|
||||
}
|
||||
if strings.TrimSpace(*csvPath) == "" || strings.TrimSpace(*tenant) == "" {
|
||||
return errors.New("-csv and -tenant are required")
|
||||
}
|
||||
input, err := os.Open(*csvPath)
|
||||
if err != nil {
|
||||
return errors.New("open directory CSV failed")
|
||||
}
|
||||
defer input.Close()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
if *validateOnly {
|
||||
count, importErr := directory.ImportCSV(ctx, *tenant, input, validationWriter{})
|
||||
if importErr != nil {
|
||||
return errors.New("directory CSV validation failed")
|
||||
}
|
||||
_, _ = fmt.Fprintf(out, "validated_profiles=%d\n", count)
|
||||
return nil
|
||||
}
|
||||
if strings.TrimSpace(*databaseURLFile) == "" {
|
||||
return errors.New("-database-url-file is required for import")
|
||||
}
|
||||
databaseURL, err := readSecretFile(*databaseURLFile)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
store, err := postgres.Open(ctx, databaseURL)
|
||||
if err != nil {
|
||||
return errors.New("open directory database failed")
|
||||
}
|
||||
defer store.Close()
|
||||
count, err := directory.ImportCSV(ctx, *tenant, input, store)
|
||||
if err != nil {
|
||||
return errors.New("directory CSV import failed")
|
||||
}
|
||||
_, _ = fmt.Fprintf(out, "imported_profiles=%d\n", count)
|
||||
return nil
|
||||
}
|
||||
|
||||
type validationWriter struct{}
|
||||
|
||||
func (validationWriter) UpsertDirectoryProfiles(context.Context, string, []directory.Profile) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func readSecretFile(path string) (string, error) {
|
||||
raw, err := os.ReadFile(strings.TrimSpace(path))
|
||||
if err != nil { return "", errors.New("read database URL file failed") }
|
||||
if len(raw) > 16<<10 { return "", errors.New("database URL file is too large") }
|
||||
value := strings.TrimSpace(string(raw))
|
||||
if value == "" { return "", errors.New("database URL file is empty") }
|
||||
return value, nil
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestReadSecretFileTrimsValue(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "database_url")
|
||||
if err := os.WriteFile(path, []byte(" postgres://secret-value \n"), 0o600); err != nil { t.Fatal(err) }
|
||||
value, err := readSecretFile(path)
|
||||
if err != nil { t.Fatalf("readSecretFile: %v", err) }
|
||||
if value != "postgres://secret-value" { t.Fatalf("value=%q", value) }
|
||||
}
|
||||
|
||||
func TestReadSecretFileErrorDoesNotLeakContents(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "empty")
|
||||
if err := os.WriteFile(path, []byte(""), 0o600); err != nil { t.Fatal(err) }
|
||||
_, err := readSecretFile(path)
|
||||
if err == nil { t.Fatal("expected error") }
|
||||
if strings.Contains(err.Error(), "secret-value") { t.Fatalf("error leaked secret: %v", err) }
|
||||
}
|
||||
|
||||
func TestRunValidateOnlyDoesNotRequireDatabase(t *testing.T) {
|
||||
csvPath := filepath.Join(t.TempDir(), "directory.csv")
|
||||
if err := os.WriteFile(csvPath, []byte("user_id,union_id,name,nickname,title,department_ids,dept_name\nstaff-1,,张三,,工程师,100,技术支持部\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var out bytes.Buffer
|
||||
err := run([]string{"-csv", csvPath, "-tenant", "tenant-a", "-validate-only"}, &out)
|
||||
if err != nil {
|
||||
t.Fatalf("run validate-only: %v", err)
|
||||
}
|
||||
if !strings.Contains(out.String(), "validated_profiles=1") {
|
||||
t.Fatalf("output=%q", out.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunRequiresCSVAndDatabaseURLFileBeforeConnecting(t *testing.T) {
|
||||
var out bytes.Buffer
|
||||
for _, args := range [][]string{{}, {"-csv", "snapshot.csv"}} {
|
||||
err := run(args, &out)
|
||||
if err == nil { t.Fatalf("run(%v) expected error", args) }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestMainWiresDirectoryResolverIntoDingTalkHandler(t *testing.T) {
|
||||
raw, err := os.ReadFile("main.go")
|
||||
if err != nil {
|
||||
t.Fatalf("read main.go: %v", err)
|
||||
}
|
||||
src := string(raw)
|
||||
for _, required := range []string{
|
||||
"directory.NewDingTalkLookup",
|
||||
"directory.NewResolver",
|
||||
"Directory: directoryResolver",
|
||||
} {
|
||||
if !strings.Contains(src, required) {
|
||||
t.Errorf("main.go missing %q", required)
|
||||
}
|
||||
}
|
||||
}
|
||||
+16
-1
@@ -15,6 +15,7 @@ import (
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/audit"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/channels/dingtalk"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/coverage"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/directory"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/feedback"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/httpapi"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/qa"
|
||||
@@ -100,6 +101,20 @@ func main() {
|
||||
}
|
||||
var streamClient *dingtalk.StreamClient
|
||||
if dingCfg.Enabled() {
|
||||
directoryLookup, err := directory.NewDingTalkLookup(directory.DingTalkLookupConfig{
|
||||
TokenURL: "https://oapi.dingtalk.com/gettoken",
|
||||
APIBaseURL: "https://oapi.dingtalk.com",
|
||||
ClientID: dingCfg.ClientID,
|
||||
ClientSecret: dingCfg.ClientSecret,
|
||||
Timeout: 10 * time.Second,
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatalf("DingTalk directory lookup init failed: %v", err)
|
||||
}
|
||||
directoryResolver, err := directory.NewResolver(store, directoryLookup)
|
||||
if err != nil {
|
||||
log.Fatalf("DingTalk directory resolver init failed: %v", err)
|
||||
}
|
||||
feedbackService, err := feedback.NewService(dingCfg.TenantKey, "dingtalk", store)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
@@ -138,7 +153,7 @@ func main() {
|
||||
dingtalk.TenantConfig{TenantKey: dingCfg.TenantKey, Channel: "dingtalk"},
|
||||
qaService,
|
||||
sender,
|
||||
dingtalk.HandlerOptions{ImageDownloader: downloader, CardSender: cardSender},
|
||||
dingtalk.HandlerOptions{ImageDownloader: downloader, CardSender: cardSender, Directory: directoryResolver},
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
ALTER TABLE session_mappings
|
||||
DROP COLUMN IF EXISTS display_title;
|
||||
@@ -0,0 +1,2 @@
|
||||
ALTER TABLE session_mappings
|
||||
ADD COLUMN IF NOT EXISTS display_title TEXT NOT NULL DEFAULT '';
|
||||
@@ -0,0 +1,28 @@
|
||||
package migrations
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSessionDisplayTitleMigration(t *testing.T) {
|
||||
up, err := os.ReadFile(filepath.Join("000008_session_display_title.up.sql"))
|
||||
if err != nil {
|
||||
t.Fatalf("read up migration: %v", err)
|
||||
}
|
||||
down, err := os.ReadFile(filepath.Join("000008_session_display_title.down.sql"))
|
||||
if err != nil {
|
||||
t.Fatalf("read down migration: %v", err)
|
||||
}
|
||||
upper := strings.ToUpper(string(up))
|
||||
for _, required := range []string{"ALTER TABLE SESSION_MAPPINGS", "ADD COLUMN IF NOT EXISTS DISPLAY_TITLE", "TEXT NOT NULL", "DEFAULT ''"} {
|
||||
if !strings.Contains(upper, required) {
|
||||
t.Errorf("up migration missing %q", required)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(strings.ToUpper(string(down)), "DROP COLUMN IF EXISTS DISPLAY_TITLE") {
|
||||
t.Error("down migration must drop display_title")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
ALTER TABLE messages
|
||||
DROP COLUMN IF EXISTS region_snapshot;
|
||||
|
||||
ALTER TABLE messages
|
||||
DROP COLUMN IF EXISTS department_snapshot;
|
||||
|
||||
DROP TABLE IF EXISTS directory_profiles;
|
||||
@@ -0,0 +1,26 @@
|
||||
CREATE TABLE IF NOT EXISTS directory_profiles (
|
||||
tenant_key TEXT NOT NULL,
|
||||
staff_id TEXT NOT NULL,
|
||||
union_id TEXT NOT NULL DEFAULT '',
|
||||
name TEXT NOT NULL,
|
||||
nickname TEXT NOT NULL DEFAULT '',
|
||||
job_title TEXT NOT NULL DEFAULT '',
|
||||
department_ids JSONB NOT NULL DEFAULT '[]'::JSONB,
|
||||
department_name TEXT NOT NULL DEFAULT '',
|
||||
main_department_name TEXT NOT NULL DEFAULT '',
|
||||
region TEXT NOT NULL DEFAULT '',
|
||||
employment_status TEXT NOT NULL DEFAULT '',
|
||||
source TEXT NOT NULL CHECK (source IN ('full_sync','on_demand')),
|
||||
synced_at TIMESTAMPTZ NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
PRIMARY KEY (tenant_key, staff_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_directory_profiles_tenant_name
|
||||
ON directory_profiles (tenant_key, name);
|
||||
|
||||
ALTER TABLE messages
|
||||
ADD COLUMN IF NOT EXISTS department_snapshot TEXT NOT NULL DEFAULT '';
|
||||
|
||||
ALTER TABLE messages
|
||||
ADD COLUMN IF NOT EXISTS region_snapshot TEXT NOT NULL DEFAULT '';
|
||||
@@ -0,0 +1,42 @@
|
||||
package migrations
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestDirectoryProfilesMigration(t *testing.T) {
|
||||
up, err := os.ReadFile(filepath.Join("000009_directory_profiles.up.sql"))
|
||||
if err != nil {
|
||||
t.Fatalf("read up migration: %v", err)
|
||||
}
|
||||
down, err := os.ReadFile(filepath.Join("000009_directory_profiles.down.sql"))
|
||||
if err != nil {
|
||||
t.Fatalf("read down migration: %v", err)
|
||||
}
|
||||
upper := strings.ToUpper(string(up))
|
||||
for _, required := range []string{
|
||||
"CREATE TABLE IF NOT EXISTS DIRECTORY_PROFILES",
|
||||
"PRIMARY KEY (TENANT_KEY, STAFF_ID)",
|
||||
"DEPARTMENT_IDS JSONB",
|
||||
"REGION TEXT",
|
||||
"SOURCE TEXT",
|
||||
"ADD COLUMN IF NOT EXISTS DEPARTMENT_SNAPSHOT",
|
||||
"ADD COLUMN IF NOT EXISTS REGION_SNAPSHOT",
|
||||
} {
|
||||
if !strings.Contains(upper, required) {
|
||||
t.Errorf("up migration missing %q", required)
|
||||
}
|
||||
}
|
||||
for _, forbidden := range []string{"MOBILE", "EMAIL", "CLIENT_SECRET", "ACCESS_TOKEN", "RAW_PAYLOAD"} {
|
||||
if strings.Contains(upper, forbidden) {
|
||||
t.Errorf("up migration contains forbidden sensitive field %q", forbidden)
|
||||
}
|
||||
}
|
||||
downUpper := strings.ToUpper(string(down))
|
||||
if !strings.Contains(downUpper, "DROP TABLE IF EXISTS DIRECTORY_PROFILES") || !strings.Contains(downUpper, "DROP COLUMN IF EXISTS REGION_SNAPSHOT") {
|
||||
t.Error("down migration must remove directory table and snapshot columns")
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"time"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/directory"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
|
||||
)
|
||||
|
||||
@@ -32,17 +33,18 @@ type TenantConfig struct {
|
||||
}
|
||||
|
||||
type BotCallback struct {
|
||||
ConversationType string
|
||||
ConversationID string
|
||||
Msgtype string
|
||||
MsgID string
|
||||
SenderNick string
|
||||
SenderStaffID string
|
||||
SenderID string
|
||||
IsInAtList bool
|
||||
SessionWebhook string
|
||||
Text string
|
||||
DownloadCode string
|
||||
ConversationType string
|
||||
ConversationID string
|
||||
ConversationTitle string
|
||||
Msgtype string
|
||||
MsgID string
|
||||
SenderNick string
|
||||
SenderStaffID string
|
||||
SenderID string
|
||||
IsInAtList bool
|
||||
SessionWebhook string
|
||||
Text string
|
||||
DownloadCode string
|
||||
}
|
||||
|
||||
type QAAsker interface {
|
||||
@@ -73,6 +75,10 @@ type ImageDownloader interface {
|
||||
Download(ctx context.Context, downloadCode string) (core.ImageInput, error)
|
||||
}
|
||||
|
||||
type UserDirectory interface {
|
||||
ResolveUser(ctx context.Context, tenantKey, staffID, fallbackName string) (directory.Profile, error)
|
||||
}
|
||||
|
||||
// FinalCardSender sends an AI card as the final reply. Implementations must
|
||||
// not log err, outTrackID, or content at any level.
|
||||
type FinalCardSender interface {
|
||||
@@ -89,6 +95,7 @@ type StreamingCardSender interface {
|
||||
type HandlerOptions struct {
|
||||
ImageDownloader ImageDownloader
|
||||
CardSender FinalCardSender
|
||||
Directory UserDirectory
|
||||
}
|
||||
|
||||
type Handler struct {
|
||||
@@ -97,6 +104,7 @@ type Handler struct {
|
||||
reply MarkdownSender
|
||||
downloader ImageDownloader
|
||||
card FinalCardSender
|
||||
directory UserDirectory
|
||||
now func() time.Time
|
||||
cardUpdateInterval time.Duration
|
||||
finalDeliveryTimeout time.Duration
|
||||
@@ -137,6 +145,7 @@ func NewHandlerWithOptions(cfg TenantConfig, qa QAAsker, reply MarkdownSender, o
|
||||
reply: reply,
|
||||
downloader: opts.ImageDownloader,
|
||||
card: opts.CardSender,
|
||||
directory: opts.Directory,
|
||||
now: time.Now,
|
||||
cardUpdateInterval: 500 * time.Millisecond,
|
||||
finalDeliveryTimeout: defaultHTTPTimeout,
|
||||
@@ -170,13 +179,28 @@ func (h *Handler) Handle(ctx context.Context, cb BotCallback) error {
|
||||
return errors.New("dingtalk handler: sender staffId/senderId is required")
|
||||
}
|
||||
|
||||
senderName := strings.TrimSpace(cb.SenderNick)
|
||||
department := ""
|
||||
region := ""
|
||||
if h.directory != nil {
|
||||
profile, _ := h.directory.ResolveUser(ctx, h.cfg.TenantKey, externalUserID, senderName)
|
||||
if name := strings.TrimSpace(profile.Name); name != "" {
|
||||
senderName = name
|
||||
department = strings.TrimSpace(profile.DepartmentName)
|
||||
region = strings.TrimSpace(profile.Region)
|
||||
}
|
||||
}
|
||||
|
||||
req := core.QARequest{
|
||||
TenantKey: h.cfg.TenantKey,
|
||||
Channel: h.cfg.Channel,
|
||||
ExternalUserID: externalUserID,
|
||||
ExternalMessageID: strings.TrimSpace(cb.MsgID),
|
||||
SenderName: cb.SenderNick,
|
||||
SenderName: senderName,
|
||||
Department: department,
|
||||
Region: region,
|
||||
MessageType: strings.ToLower(strings.TrimSpace(cb.Msgtype)),
|
||||
ConversationTitle: strings.TrimSpace(cb.ConversationTitle),
|
||||
}
|
||||
switch cb.ConversationType {
|
||||
case conversationTypeGroup:
|
||||
@@ -187,6 +211,7 @@ func (h *Handler) Handle(ctx context.Context, cb BotCallback) error {
|
||||
req.ExternalChatID = strings.TrimSpace(cb.ConversationID)
|
||||
default:
|
||||
req.ConversationType = core.ConversationDirect
|
||||
req.ConversationTitle = ""
|
||||
}
|
||||
|
||||
switch strings.ToLower(strings.TrimSpace(cb.Msgtype)) {
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"time"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/directory"
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
|
||||
)
|
||||
|
||||
@@ -31,6 +32,17 @@ func (f *fakeQA) RecordDelivery(_ context.Context, update core.DeliveryUpdate) e
|
||||
return f.deliveryErr
|
||||
}
|
||||
|
||||
type fakeDirectory struct {
|
||||
profile directory.Profile
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeDirectory) ResolveUser(context.Context, string, string, string) (directory.Profile, error) {
|
||||
f.calls++
|
||||
return f.profile, f.err
|
||||
}
|
||||
|
||||
type fakeReply struct {
|
||||
called bool
|
||||
lastURL string
|
||||
@@ -192,6 +204,42 @@ func TestHandleDirectTextCallsQAAndRepliesAnswer(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleDirectUsesDirectoryProfile(t *testing.T) {
|
||||
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
|
||||
reply := &fakeReply{}
|
||||
profiles := &fakeDirectory{profile: directory.Profile{Name: "张三", DepartmentName: "技术支持部", Region: "APAC"}}
|
||||
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{Directory: profiles})
|
||||
if err != nil {
|
||||
t.Fatalf("NewHandlerWithOptions: %v", err)
|
||||
}
|
||||
data := directTextCallback()
|
||||
data.SenderNick = "钉钉昵称"
|
||||
if err := h.Handle(context.Background(), data); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if profiles.calls != 1 || qa.lastReq.SenderName != "张三" || qa.lastReq.Department != "技术支持部" || qa.lastReq.Region != "APAC" {
|
||||
t.Fatalf("calls=%d request=%+v", profiles.calls, qa.lastReq)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleDirectoryFailureFallsBackToSenderNick(t *testing.T) {
|
||||
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
|
||||
reply := &fakeReply{}
|
||||
profiles := &fakeDirectory{err: errors.New("directory unavailable")}
|
||||
h, err := NewHandlerWithOptions(TenantConfig{TenantKey: "tenant-a", Channel: "dingtalk"}, qa, reply, HandlerOptions{Directory: profiles})
|
||||
if err != nil {
|
||||
t.Fatalf("NewHandlerWithOptions: %v", err)
|
||||
}
|
||||
data := directTextCallback()
|
||||
data.SenderNick = " 回调昵称 "
|
||||
if err := h.Handle(context.Background(), data); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if qa.lastReq.SenderName != "回调昵称" || qa.lastReq.Department != "" || qa.lastReq.Region != "" {
|
||||
t.Fatalf("request=%+v", qa.lastReq)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleDirectTextFallsBackToSenderID(t *testing.T) {
|
||||
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
|
||||
reply := &fakeReply{}
|
||||
@@ -1796,3 +1844,33 @@ func TestDeliveryAuditEventStreamingProductionPath(t *testing.T) {
|
||||
}
|
||||
requireDelivery(t, qa.deliveries, core.DeliverySent, core.DeliveryMethodCard, "event-track-1", "")
|
||||
}
|
||||
|
||||
func TestHandlePassesConversationTitleToQARequest(t *testing.T) {
|
||||
qa := &fakeQA{result: core.QAResult{Answer: "ok"}}
|
||||
reply := &fakeReply{}
|
||||
h := newTestHandler(t, qa, reply)
|
||||
|
||||
t.Run("group conversation title propagated", func(t *testing.T) {
|
||||
cb := groupTextCallback()
|
||||
cb.ConversationTitle = "项目讨论群"
|
||||
if err := h.Handle(context.Background(), cb); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if qa.lastReq.ConversationTitle != "项目讨论群" {
|
||||
t.Fatalf("ConversationTitle = %q, want 项目讨论群", qa.lastReq.ConversationTitle)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("direct conversation title is empty", func(t *testing.T) {
|
||||
qa2 := &fakeQA{result: core.QAResult{Answer: "ok"}}
|
||||
h2 := newTestHandler(t, qa2, &fakeReply{})
|
||||
cb := directTextCallback()
|
||||
cb.ConversationTitle = "should not appear"
|
||||
if err := h2.Handle(context.Background(), cb); err != nil {
|
||||
t.Fatalf("Handle: %v", err)
|
||||
}
|
||||
if qa2.lastReq.ConversationTitle != "" {
|
||||
t.Fatalf("ConversationTitle = %q, want empty for direct", qa2.lastReq.ConversationTitle)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -69,16 +69,17 @@ func BotCallbackFromSDK(data *chatbot.BotCallbackDataModel) (BotCallback, error)
|
||||
}
|
||||
msgtype := strings.ToLower(strings.TrimSpace(data.Msgtype))
|
||||
cb := BotCallback{
|
||||
ConversationType: data.ConversationType,
|
||||
ConversationID: data.ConversationId,
|
||||
Msgtype: msgtype,
|
||||
MsgID: data.MsgId,
|
||||
SenderNick: data.SenderNick,
|
||||
SenderStaffID: data.SenderStaffId,
|
||||
SenderID: data.SenderId,
|
||||
IsInAtList: data.IsInAtList,
|
||||
SessionWebhook: data.SessionWebhook,
|
||||
Text: data.Text.Content,
|
||||
ConversationType: data.ConversationType,
|
||||
ConversationID: data.ConversationId,
|
||||
ConversationTitle: data.ConversationTitle,
|
||||
Msgtype: msgtype,
|
||||
MsgID: data.MsgId,
|
||||
SenderNick: data.SenderNick,
|
||||
SenderStaffID: data.SenderStaffId,
|
||||
SenderID: data.SenderId,
|
||||
IsInAtList: data.IsInAtList,
|
||||
SessionWebhook: data.SessionWebhook,
|
||||
Text: data.Text.Content,
|
||||
}
|
||||
if msgtype == "richtext" {
|
||||
rtMsgtype, rtText, rtCode := parseRichText(data.Content)
|
||||
|
||||
@@ -542,6 +542,46 @@ func TestBotCallbackFromSDKRichText(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestBotCallbackFromSDKConversationTitle(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
t.Run("conversation title survives conversion", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
data := &chatbot.BotCallbackDataModel{
|
||||
ConversationType: "2",
|
||||
ConversationId: "cid123",
|
||||
Msgtype: "text",
|
||||
MsgId: "msg-1",
|
||||
ConversationTitle: "项目讨论群",
|
||||
Text: chatbot.BotCallbackDataTextModel{Content: "hello"},
|
||||
}
|
||||
cb, err := BotCallbackFromSDK(data)
|
||||
if err != nil {
|
||||
t.Fatalf("BotCallbackFromSDK() error = %v", err)
|
||||
}
|
||||
if cb.ConversationTitle != "项目讨论群" {
|
||||
t.Fatalf("ConversationTitle = %q, want 项目讨论群", cb.ConversationTitle)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("conversation title empty for direct", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
data := &chatbot.BotCallbackDataModel{
|
||||
ConversationType: "1",
|
||||
Msgtype: "text",
|
||||
MsgId: "msg-2",
|
||||
Text: chatbot.BotCallbackDataTextModel{Content: "hi"},
|
||||
}
|
||||
cb, err := BotCallbackFromSDK(data)
|
||||
if err != nil {
|
||||
t.Fatalf("BotCallbackFromSDK() error = %v", err)
|
||||
}
|
||||
if cb.ConversationTitle != "" {
|
||||
t.Fatalf("ConversationTitle = %q, want empty for direct", cb.ConversationTitle)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestNewStreamClientValidation(t *testing.T) {
|
||||
t.Parallel()
|
||||
h := mustTestHandler(t)
|
||||
|
||||
@@ -40,10 +40,13 @@ type QARequest struct {
|
||||
TenantKey string
|
||||
Channel string
|
||||
ConversationType ConversationType
|
||||
ConversationTitle string
|
||||
ExternalChatID string
|
||||
ExternalUserID string
|
||||
ExternalMessageID string
|
||||
SenderName string
|
||||
Department string
|
||||
Region string
|
||||
Query string
|
||||
Images []ImageInput
|
||||
MessageType string
|
||||
@@ -77,8 +80,11 @@ type SessionMapping struct {
|
||||
ConversationType ConversationType
|
||||
ExternalSubjectID string
|
||||
WeKnoraSessionID string
|
||||
DisplayTitle string
|
||||
}
|
||||
|
||||
type SessionTitleUpdate func(ctx context.Context, sessionID, title string) error
|
||||
|
||||
type Turn struct {
|
||||
ID string
|
||||
TenantKey string
|
||||
@@ -89,9 +95,12 @@ type Turn struct {
|
||||
WeKnoraRequestID string
|
||||
AgentID string
|
||||
ConversationType ConversationType
|
||||
ConversationTitle string
|
||||
ExternalChatID string
|
||||
ExternalUserID string
|
||||
SenderName string
|
||||
DepartmentSnapshot string
|
||||
RegionSnapshot string
|
||||
MessageType string
|
||||
ActorID string
|
||||
ConversationID string
|
||||
@@ -143,6 +152,7 @@ var ErrDeliveryUpdateFailed = errors.New("delivery update failed: RowsAffected !
|
||||
type Repository interface {
|
||||
BeginTurn(ctx context.Context, turn Turn) (Turn, bool, error)
|
||||
ResolveSession(ctx context.Context, mapping SessionMapping, create func(context.Context) (string, error)) (string, error)
|
||||
SyncSessionTitle(ctx context.Context, mapping SessionMapping, title string, update SessionTitleUpdate) error
|
||||
FinishTurn(ctx context.Context, turn Turn) error
|
||||
RecordDelivery(ctx context.Context, du DeliveryUpdate) error
|
||||
}
|
||||
|
||||
@@ -0,0 +1,121 @@
|
||||
package directory
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type DingTalkLookupConfig struct {
|
||||
TokenURL string
|
||||
APIBaseURL string
|
||||
ClientID string
|
||||
ClientSecret string
|
||||
Timeout time.Duration
|
||||
}
|
||||
|
||||
type DingTalkLookup struct {
|
||||
tokenURL string
|
||||
apiBaseURL string
|
||||
clientID string
|
||||
clientSecret string
|
||||
httpClient *http.Client
|
||||
mu sync.Mutex
|
||||
token string
|
||||
tokenExpiry time.Time
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func NewDingTalkLookup(cfg DingTalkLookupConfig) (*DingTalkLookup, error) {
|
||||
cfg.TokenURL = strings.TrimSpace(cfg.TokenURL)
|
||||
cfg.APIBaseURL = strings.TrimRight(strings.TrimSpace(cfg.APIBaseURL), "/")
|
||||
if cfg.TokenURL == "" || cfg.APIBaseURL == "" || strings.TrimSpace(cfg.ClientID) == "" || strings.TrimSpace(cfg.ClientSecret) == "" {
|
||||
return nil, errors.New("dingtalk lookup endpoints and credentials are required")
|
||||
}
|
||||
if _, err := url.ParseRequestURI(cfg.TokenURL); err != nil { return nil, errors.New("invalid dingtalk token endpoint") }
|
||||
if _, err := url.ParseRequestURI(cfg.APIBaseURL); err != nil { return nil, errors.New("invalid dingtalk API endpoint") }
|
||||
if cfg.Timeout <= 0 { cfg.Timeout = 10 * time.Second }
|
||||
return &DingTalkLookup{
|
||||
tokenURL: cfg.TokenURL, apiBaseURL: cfg.APIBaseURL,
|
||||
clientID: cfg.ClientID, clientSecret: cfg.ClientSecret,
|
||||
httpClient: &http.Client{Timeout: cfg.Timeout}, now: time.Now,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *DingTalkLookup) LookupUser(ctx context.Context, staffID string) (Profile, error) {
|
||||
staffID = strings.TrimSpace(staffID)
|
||||
if staffID == "" { return Profile{}, errors.New("dingtalk user id is required") }
|
||||
token, err := c.accessToken(ctx)
|
||||
if err != nil { return Profile{}, err }
|
||||
payload, _ := json.Marshal(map[string]string{"userid": staffID})
|
||||
endpoint := c.apiBaseURL + "/topapi/v2/user/get?access_token=" + url.QueryEscape(token)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload))
|
||||
if err != nil { return Profile{}, errors.New("dingtalk user request failed") }
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil { return Profile{}, errors.New("dingtalk user request failed") }
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 { return Profile{}, errors.New("dingtalk user HTTP failure") }
|
||||
var decoded struct {
|
||||
ErrCode int `json:"errcode"`
|
||||
Result struct {
|
||||
UserID string `json:"userid"`
|
||||
UnionID string `json:"unionid"`
|
||||
Name string `json:"name"`
|
||||
Nickname string `json:"nickname"`
|
||||
Title string `json:"title"`
|
||||
DepartmentIDs []int64 `json:"dept_id_list"`
|
||||
Status int `json:"status"`
|
||||
} `json:"result"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&decoded); err != nil || decoded.ErrCode != 0 {
|
||||
return Profile{}, errors.New("dingtalk user response invalid")
|
||||
}
|
||||
ids := make([]string, 0, len(decoded.Result.DepartmentIDs))
|
||||
for _, id := range decoded.Result.DepartmentIDs { ids = append(ids, strconv.FormatInt(id, 10)) }
|
||||
return Profile{
|
||||
StaffID: staffID, UnionID: decoded.Result.UnionID, Name: strings.TrimSpace(decoded.Result.Name),
|
||||
Nickname: decoded.Result.Nickname, JobTitle: decoded.Result.Title, DepartmentIDs: ids,
|
||||
EmploymentStatus: strconv.Itoa(decoded.Result.Status), Source: SourceOnDemand, SyncedAt: c.now().UTC(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *DingTalkLookup) accessToken(ctx context.Context) (string, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.token != "" && c.now().Before(c.tokenExpiry) { return c.token, nil }
|
||||
parsed, err := url.Parse(c.tokenURL)
|
||||
if err != nil { return "", errors.New("dingtalk token request failed") }
|
||||
query := parsed.Query()
|
||||
query.Set("appkey", c.clientID)
|
||||
query.Set("appsecret", c.clientSecret)
|
||||
parsed.RawQuery = query.Encode()
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, parsed.String(), nil)
|
||||
if err != nil { return "", errors.New("dingtalk token request failed") }
|
||||
resp, err := c.httpClient.Do(req)
|
||||
if err != nil { return "", errors.New("dingtalk token request failed") }
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 { return "", errors.New("dingtalk token HTTP failure") }
|
||||
var decoded struct {
|
||||
ErrCode int `json:"errcode"`
|
||||
AccessToken string `json:"access_token"`
|
||||
ExpiresIn int `json:"expires_in"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&decoded); err != nil || decoded.ErrCode != 0 || decoded.AccessToken == "" {
|
||||
return "", errors.New("dingtalk token response invalid")
|
||||
}
|
||||
ttl := time.Duration(decoded.ExpiresIn) * time.Second
|
||||
if ttl <= 0 { ttl = time.Hour }
|
||||
refreshMargin := 5 * time.Minute
|
||||
if ttl <= refreshMargin { refreshMargin = ttl / 2 }
|
||||
c.token = decoded.AccessToken
|
||||
c.tokenExpiry = c.now().Add(ttl - refreshMargin)
|
||||
return c.token, nil
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package directory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestDingTalkLookupFetchesUserAndCachesToken(t *testing.T) {
|
||||
var tokenCalls atomic.Int32
|
||||
var userCalls atomic.Int32
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/gettoken", func(w http.ResponseWriter, r *http.Request) {
|
||||
tokenCalls.Add(1)
|
||||
if r.Method != http.MethodGet || r.URL.Query().Get("appkey") != "app-key" || r.URL.Query().Get("appsecret") != "app-secret" {
|
||||
t.Fatalf("unexpected token request: %s %s", r.Method, r.URL.String())
|
||||
}
|
||||
_, _ = w.Write([]byte(`{"errcode":0,"access_token":"token-secret","expires_in":7200}`))
|
||||
})
|
||||
mux.HandleFunc("/topapi/v2/user/get", func(w http.ResponseWriter, r *http.Request) {
|
||||
userCalls.Add(1)
|
||||
if r.Method != http.MethodPost || r.URL.Query().Get("access_token") != "token-secret" {
|
||||
t.Fatalf("unexpected user request: %s %s", r.Method, r.URL.String())
|
||||
}
|
||||
var body [256]byte
|
||||
n, _ := r.Body.Read(body[:])
|
||||
if !strings.Contains(string(body[:n]), `"userid":"staff-1"`) {
|
||||
t.Fatalf("unexpected body: %s", body[:n])
|
||||
}
|
||||
_, _ = w.Write([]byte(`{"errcode":0,"result":{"userid":"staff-1","unionid":"union-1","name":"张三","nickname":"小张","title":"工程师","dept_id_list":[100,200],"status":5}}`))
|
||||
})
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
lookup, err := NewDingTalkLookup(DingTalkLookupConfig{
|
||||
TokenURL: server.URL + "/gettoken", APIBaseURL: server.URL,
|
||||
ClientID: "app-key", ClientSecret: "app-secret", Timeout: 2 * time.Second,
|
||||
})
|
||||
if err != nil { t.Fatal(err) }
|
||||
for i := 0; i < 2; i++ {
|
||||
profile, err := lookup.LookupUser(context.Background(), "staff-1")
|
||||
if err != nil { t.Fatalf("LookupUser(%d): %v", i, err) }
|
||||
if profile.Name != "张三" || profile.UnionID != "union-1" || profile.JobTitle != "工程师" || len(profile.DepartmentIDs) != 2 || profile.DepartmentIDs[1] != "200" {
|
||||
t.Fatalf("profile=%+v", profile)
|
||||
}
|
||||
}
|
||||
if tokenCalls.Load() != 1 || userCalls.Load() != 2 {
|
||||
t.Fatalf("token calls=%d user calls=%d", tokenCalls.Load(), userCalls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestDingTalkLookupErrorsAreSanitized(t *testing.T) {
|
||||
for _, failPath := range []string{"token", "user"} {
|
||||
t.Run(failPath, func(t *testing.T) {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/gettoken", func(w http.ResponseWriter, _ *http.Request) {
|
||||
if failPath == "token" { http.Error(w, "raw-secret-body", 500); return }
|
||||
_, _ = w.Write([]byte(`{"errcode":0,"access_token":"token-secret","expires_in":7200}`))
|
||||
})
|
||||
mux.HandleFunc("/topapi/v2/user/get", func(w http.ResponseWriter, _ *http.Request) {
|
||||
http.Error(w, "raw-secret-body", 500)
|
||||
})
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
lookup, err := NewDingTalkLookup(DingTalkLookupConfig{TokenURL: server.URL + "/gettoken", APIBaseURL: server.URL, ClientID: "app-key", ClientSecret: "app-secret", Timeout: time.Second})
|
||||
if err != nil { t.Fatal(err) }
|
||||
_, err = lookup.LookupUser(context.Background(), "staff-secret")
|
||||
if err == nil { t.Fatal("expected error") }
|
||||
for _, forbidden := range []string{"raw-secret-body", "app-secret", "token-secret", "staff-secret"} {
|
||||
if strings.Contains(err.Error(), forbidden) { t.Fatalf("error leaked %q: %v", forbidden, err) }
|
||||
}
|
||||
if !strings.Contains(err.Error(), fmt.Sprintf("dingtalk %s", failPath)) {
|
||||
t.Fatalf("error=%q", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
package directory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/csv"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
var requiredCSVHeaders = []string{"user_id", "name", "dept_name"}
|
||||
|
||||
// ImportCSV validates the complete snapshot before making one repository call.
|
||||
// It consumes the tested Python sync output and accepts optional production
|
||||
// enrichment columns without reimplementing the bulk DingTalk crawler in Go.
|
||||
func ImportCSV(ctx context.Context, tenantKey string, input io.Reader, writer SnapshotWriter) (int, error) {
|
||||
tenantKey = strings.TrimSpace(tenantKey)
|
||||
if tenantKey == "" || input == nil || writer == nil {
|
||||
return 0, errors.New("tenant, CSV input and writer are required")
|
||||
}
|
||||
r := csv.NewReader(input)
|
||||
r.FieldsPerRecord = -1
|
||||
header, err := r.Read()
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("read directory CSV header: %w", err)
|
||||
}
|
||||
columns := make(map[string]int, len(header))
|
||||
for i, name := range header {
|
||||
name = strings.TrimSpace(strings.TrimPrefix(name, "\ufeff"))
|
||||
columns[name] = i
|
||||
}
|
||||
for _, name := range requiredCSVHeaders {
|
||||
if _, ok := columns[name]; !ok {
|
||||
return 0, fmt.Errorf("directory CSV missing required header %q", name)
|
||||
}
|
||||
}
|
||||
|
||||
profiles := make([]Profile, 0, 256)
|
||||
line := 1
|
||||
for {
|
||||
record, readErr := r.Read()
|
||||
if errors.Is(readErr, io.EOF) {
|
||||
break
|
||||
}
|
||||
line++
|
||||
if readErr != nil {
|
||||
return 0, fmt.Errorf("read directory CSV line %d: %w", line, readErr)
|
||||
}
|
||||
value := func(name string) string {
|
||||
idx, ok := columns[name]
|
||||
if !ok || idx >= len(record) {
|
||||
return ""
|
||||
}
|
||||
return strings.TrimSpace(record[idx])
|
||||
}
|
||||
staffID := value("user_id")
|
||||
name := value("name")
|
||||
if staffID == "" || name == "" {
|
||||
return 0, fmt.Errorf("directory CSV line %d has empty user_id or name", line)
|
||||
}
|
||||
profiles = append(profiles, Profile{
|
||||
StaffID: staffID,
|
||||
UnionID: value("union_id"),
|
||||
Name: name,
|
||||
Nickname: value("nickname"),
|
||||
JobTitle: value("title"),
|
||||
DepartmentIDs: splitDepartmentIDs(value("department_ids")),
|
||||
DepartmentName: value("dept_name"),
|
||||
MainDepartmentName: value("main_department_name"),
|
||||
Region: value("region"),
|
||||
EmploymentStatus: value("employment_status"),
|
||||
Source: SourceFullSync,
|
||||
SyncedAt: parseSyncTime(value("sync_time")),
|
||||
})
|
||||
}
|
||||
if len(profiles) == 0 {
|
||||
return 0, errors.New("directory CSV snapshot is empty")
|
||||
}
|
||||
if err := writer.UpsertDirectoryProfiles(ctx, tenantKey, profiles); err != nil {
|
||||
return 0, fmt.Errorf("write directory snapshot: %w", err)
|
||||
}
|
||||
return len(profiles), nil
|
||||
}
|
||||
|
||||
func splitDepartmentIDs(raw string) []string {
|
||||
parts := strings.FieldsFunc(raw, func(r rune) bool { return r == '/' || r == ',' || r == ';' })
|
||||
out := make([]string, 0, len(parts))
|
||||
for _, part := range parts {
|
||||
if part = strings.TrimSpace(part); part != "" {
|
||||
out = append(out, part)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func parseSyncTime(raw string) time.Time {
|
||||
for _, layout := range []string{time.RFC3339, "2006-01-02 15:04:05"} {
|
||||
if parsed, err := time.Parse(layout, raw); err == nil {
|
||||
return parsed
|
||||
}
|
||||
}
|
||||
return time.Time{}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
package directory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type recordingWriter struct {
|
||||
calls int
|
||||
tenant string
|
||||
profiles []Profile
|
||||
err error
|
||||
}
|
||||
|
||||
func (w *recordingWriter) UpsertDirectoryProfiles(_ context.Context, tenant string, profiles []Profile) error {
|
||||
w.calls++
|
||||
w.tenant = tenant
|
||||
w.profiles = append([]Profile(nil), profiles...)
|
||||
return w.err
|
||||
}
|
||||
|
||||
func TestImportCSVAcceptsTestedSyncOutputAndExtendedFields(t *testing.T) {
|
||||
csv := "\ufeffuser_id,union_id,name,nickname,title,department_ids,dept_name,main_department_name,region,employment_status,sync_time,mobile\n" +
|
||||
"staff-1,union-1, 张三 ,小张,工程师,100/200,公司/技术支持部,技术支持部,APAC,on_job,2026-07-22 02:30:00,13800000000\n"
|
||||
writer := &recordingWriter{}
|
||||
count, err := ImportCSV(context.Background(), "tenant-a", strings.NewReader(csv), writer)
|
||||
if err != nil {
|
||||
t.Fatalf("ImportCSV: %v", err)
|
||||
}
|
||||
if count != 1 || writer.calls != 1 || writer.tenant != "tenant-a" || len(writer.profiles) != 1 {
|
||||
t.Fatalf("count=%d calls=%d tenant=%q profiles=%d", count, writer.calls, writer.tenant, len(writer.profiles))
|
||||
}
|
||||
p := writer.profiles[0]
|
||||
if p.StaffID != "staff-1" || p.Name != "张三" || p.DepartmentName != "公司/技术支持部" || p.MainDepartmentName != "技术支持部" || p.Region != "APAC" || p.EmploymentStatus != "on_job" || p.Source != SourceFullSync {
|
||||
t.Fatalf("profile mismatch: %+v", p)
|
||||
}
|
||||
if len(p.DepartmentIDs) != 2 || p.DepartmentIDs[0] != "100" || p.DepartmentIDs[1] != "200" {
|
||||
t.Fatalf("department ids: %v", p.DepartmentIDs)
|
||||
}
|
||||
// The domain type intentionally has no mobile/email/raw payload fields.
|
||||
}
|
||||
|
||||
func TestImportCSVAcceptsCurrentSevenColumnOutput(t *testing.T) {
|
||||
csv := "user_id,union_id,name,nickname,title,department_ids,dept_name\n" +
|
||||
"staff-2,union-2,李四,,经理,300,公司/销售部\n"
|
||||
writer := &recordingWriter{}
|
||||
count, err := ImportCSV(context.Background(), "tenant-a", strings.NewReader(csv), writer)
|
||||
if err != nil {
|
||||
t.Fatalf("ImportCSV: %v", err)
|
||||
}
|
||||
if count != 1 || writer.profiles[0].Name != "李四" || writer.profiles[0].Region != "" {
|
||||
t.Fatalf("count=%d profile=%+v", count, writer.profiles[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestImportCSVRejectsInvalidSnapshotBeforeWrite(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
csv string
|
||||
}{
|
||||
{name: "missing required header", csv: "user_id,dept_name\nstaff-1,技术支持部\n"},
|
||||
{name: "empty snapshot", csv: "user_id,name,dept_name\n"},
|
||||
{name: "empty staff id", csv: "user_id,name,dept_name\n,张三,技术支持部\n"},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
writer := &recordingWriter{}
|
||||
_, err := ImportCSV(context.Background(), "tenant-a", strings.NewReader(tt.csv), writer)
|
||||
if err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
if writer.calls != 0 {
|
||||
t.Fatalf("writer called %d times before validation", writer.calls)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestImportCSVPropagatesWriterError(t *testing.T) {
|
||||
writer := &recordingWriter{err: errors.New("db unavailable")}
|
||||
_, err := ImportCSV(context.Background(), "tenant-a", strings.NewReader("user_id,name,dept_name\nstaff-1,张三,技术支持部\n"), writer)
|
||||
if err == nil || writer.calls != 1 {
|
||||
t.Fatalf("error=%v calls=%d", err, writer.calls)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package directory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
SourceFullSync = "full_sync"
|
||||
SourceOnDemand = "on_demand"
|
||||
)
|
||||
|
||||
// Profile is the minimum non-sensitive DingTalk directory record used by
|
||||
// DobotAI. It deliberately excludes mobile, email, credentials and raw payloads.
|
||||
type Profile struct {
|
||||
StaffID string
|
||||
UnionID string
|
||||
Name string
|
||||
Nickname string
|
||||
JobTitle string
|
||||
DepartmentIDs []string
|
||||
DepartmentName string
|
||||
MainDepartmentName string
|
||||
Region string
|
||||
EmploymentStatus string
|
||||
Source string
|
||||
SyncedAt time.Time
|
||||
}
|
||||
|
||||
type SnapshotWriter interface {
|
||||
UpsertDirectoryProfiles(ctx context.Context, tenantKey string, profiles []Profile) error
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
package directory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
type ProfileStore interface {
|
||||
LookupDirectoryProfile(ctx context.Context, tenantKey, staffID string) (Profile, bool, error)
|
||||
UpsertDirectoryProfiles(ctx context.Context, tenantKey string, profiles []Profile) error
|
||||
}
|
||||
|
||||
type UserLookup interface {
|
||||
LookupUser(ctx context.Context, staffID string) (Profile, error)
|
||||
}
|
||||
|
||||
type Resolver struct {
|
||||
store ProfileStore
|
||||
lookup UserLookup
|
||||
now func() time.Time
|
||||
}
|
||||
|
||||
func NewResolver(store ProfileStore, lookup UserLookup) (*Resolver, error) {
|
||||
if store == nil {
|
||||
return nil, errors.New("directory profile store is required")
|
||||
}
|
||||
return &Resolver{store: store, lookup: lookup, now: time.Now}, nil
|
||||
}
|
||||
|
||||
// ResolveUser always returns a safe callback-based fallback profile when the
|
||||
// cache or on-demand lookup is unavailable. The diagnostic error is intentionally
|
||||
// generic so identifiers and upstream payloads cannot leak into logs.
|
||||
func (r *Resolver) ResolveUser(ctx context.Context, tenantKey, staffID, fallbackName string) (Profile, error) {
|
||||
fallback := Profile{StaffID: strings.TrimSpace(staffID), Name: strings.TrimSpace(fallbackName)}
|
||||
profile, found, err := r.store.LookupDirectoryProfile(ctx, tenantKey, fallback.StaffID)
|
||||
if err != nil {
|
||||
return fallback, errors.New("directory local lookup failed")
|
||||
}
|
||||
if found && strings.TrimSpace(profile.Name) != "" {
|
||||
profile.Name = strings.TrimSpace(profile.Name)
|
||||
return profile, nil
|
||||
}
|
||||
if r.lookup == nil {
|
||||
return fallback, nil
|
||||
}
|
||||
profile, err = r.lookup.LookupUser(ctx, fallback.StaffID)
|
||||
if err != nil {
|
||||
return fallback, errors.New("directory on-demand lookup failed")
|
||||
}
|
||||
profile.StaffID = fallback.StaffID
|
||||
profile.Name = strings.TrimSpace(profile.Name)
|
||||
if profile.Name == "" {
|
||||
return fallback, errors.New("directory on-demand lookup returned no name")
|
||||
}
|
||||
profile.Source = SourceOnDemand
|
||||
if profile.SyncedAt.IsZero() {
|
||||
profile.SyncedAt = r.now().UTC()
|
||||
}
|
||||
if err := r.store.UpsertDirectoryProfiles(ctx, tenantKey, []Profile{profile}); err != nil {
|
||||
return profile, errors.New("directory on-demand cache write failed")
|
||||
}
|
||||
return profile, nil
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
package directory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type fakeStore struct {
|
||||
profile Profile
|
||||
found bool
|
||||
readErr error
|
||||
writeErr error
|
||||
writes int
|
||||
}
|
||||
|
||||
func (s *fakeStore) LookupDirectoryProfile(context.Context, string, string) (Profile, bool, error) {
|
||||
return s.profile, s.found, s.readErr
|
||||
}
|
||||
func (s *fakeStore) UpsertDirectoryProfiles(_ context.Context, _ string, profiles []Profile) error {
|
||||
s.writes++
|
||||
if len(profiles) > 0 {
|
||||
s.profile = profiles[0]
|
||||
}
|
||||
return s.writeErr
|
||||
}
|
||||
|
||||
type fakeLookup struct {
|
||||
profile Profile
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
func (l *fakeLookup) LookupUser(context.Context, string) (Profile, error) {
|
||||
l.calls++
|
||||
return l.profile, l.err
|
||||
}
|
||||
|
||||
func TestResolverUsesLocalProfileWithoutRemoteLookup(t *testing.T) {
|
||||
store := &fakeStore{profile: Profile{StaffID: "staff-1", Name: "张三", DepartmentName: "技术支持部", Region: "APAC"}, found: true}
|
||||
lookup := &fakeLookup{}
|
||||
resolver, err := NewResolver(store, lookup)
|
||||
if err != nil { t.Fatal(err) }
|
||||
profile, err := resolver.ResolveUser(context.Background(), "tenant-a", "staff-1", "昵称")
|
||||
if err != nil { t.Fatalf("ResolveUser: %v", err) }
|
||||
if profile.Name != "张三" || lookup.calls != 0 || store.writes != 0 {
|
||||
t.Fatalf("profile=%+v lookup=%d writes=%d", profile, lookup.calls, store.writes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolverLooksUpAndCachesMissingUser(t *testing.T) {
|
||||
store := &fakeStore{}
|
||||
lookup := &fakeLookup{profile: Profile{Name: "李四", DepartmentName: "销售部", Region: "EMEA"}}
|
||||
resolver, err := NewResolver(store, lookup)
|
||||
if err != nil { t.Fatal(err) }
|
||||
profile, err := resolver.ResolveUser(context.Background(), "tenant-a", "staff-2", "昵称")
|
||||
if err != nil { t.Fatalf("ResolveUser: %v", err) }
|
||||
if profile.StaffID != "staff-2" || profile.Name != "李四" || profile.Source != SourceOnDemand || lookup.calls != 1 || store.writes != 1 {
|
||||
t.Fatalf("profile=%+v lookup=%d writes=%d", profile, lookup.calls, store.writes)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolverFallsBackWithoutBlockingOnFailures(t *testing.T) {
|
||||
tests := []struct{
|
||||
name string
|
||||
store *fakeStore
|
||||
lookup *fakeLookup
|
||||
}{
|
||||
{name: "local read fails", store: &fakeStore{readErr: errors.New("db unavailable")}, lookup: &fakeLookup{}},
|
||||
{name: "remote lookup fails", store: &fakeStore{}, lookup: &fakeLookup{err: errors.New("remote unavailable")}},
|
||||
{name: "cache write fails but resolved profile survives", store: &fakeStore{writeErr: errors.New("db unavailable")}, lookup: &fakeLookup{profile: Profile{Name: "王五"}}},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
resolver, err := NewResolver(tt.store, tt.lookup)
|
||||
if err != nil { t.Fatal(err) }
|
||||
profile, resolveErr := resolver.ResolveUser(context.Background(), "tenant-a", "staff-3", " 回调昵称 ")
|
||||
if profile.Name == "" { t.Fatalf("empty fallback profile: %+v", profile) }
|
||||
if tt.name != "cache write fails but resolved profile survives" && profile.Name != "回调昵称" {
|
||||
t.Fatalf("profile name=%q, want fallback", profile.Name)
|
||||
}
|
||||
if resolveErr == nil { t.Fatal("expected diagnostic error") }
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolverWithoutRemoteUsesFallback(t *testing.T) {
|
||||
resolver, err := NewResolver(&fakeStore{}, nil)
|
||||
if err != nil { t.Fatal(err) }
|
||||
profile, err := resolver.ResolveUser(context.Background(), "tenant-a", "staff-4", "赵六")
|
||||
if err != nil || profile.Name != "赵六" || profile.StaffID != "staff-4" {
|
||||
t.Fatalf("profile=%+v err=%v", profile, err)
|
||||
}
|
||||
}
|
||||
+33
-20
@@ -20,6 +20,7 @@ type AgentResolver interface {
|
||||
|
||||
type WeKnora interface {
|
||||
CreateSession(ctx context.Context, title, description string) (string, error)
|
||||
UpdateSessionTitle(ctx context.Context, sessionID, title, description string) error
|
||||
AgentChat(ctx context.Context, sessionID string, request weknora.AgentChatRequest) (weknora.ChatResult, error)
|
||||
}
|
||||
|
||||
@@ -135,23 +136,26 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
|
||||
return core.QAResult{}, errors.New("unsupported message_type")
|
||||
}
|
||||
turn := core.Turn{
|
||||
ID: newID(),
|
||||
TenantKey: request.TenantKey,
|
||||
Channel: request.Channel,
|
||||
ExternalMessageID: request.ExternalMessageID,
|
||||
SessionKey: sessionKey,
|
||||
AgentID: agentID,
|
||||
ConversationType: request.ConversationType,
|
||||
ExternalChatID: request.ExternalChatID,
|
||||
ExternalUserID: request.ExternalUserID,
|
||||
SenderName: request.SenderName,
|
||||
MessageType: messageType,
|
||||
ActorID: newID(),
|
||||
ConversationID: newID(),
|
||||
RequestMessageID: newID(),
|
||||
Question: request.Query,
|
||||
Media: media,
|
||||
Status: "processing",
|
||||
ID: newID(),
|
||||
TenantKey: request.TenantKey,
|
||||
Channel: request.Channel,
|
||||
ExternalMessageID: request.ExternalMessageID,
|
||||
SessionKey: sessionKey,
|
||||
AgentID: agentID,
|
||||
ConversationType: request.ConversationType,
|
||||
ConversationTitle: request.ConversationTitle,
|
||||
ExternalChatID: request.ExternalChatID,
|
||||
ExternalUserID: request.ExternalUserID,
|
||||
SenderName: request.SenderName,
|
||||
DepartmentSnapshot: request.Department,
|
||||
RegionSnapshot: request.Region,
|
||||
MessageType: messageType,
|
||||
ActorID: newID(),
|
||||
ConversationID: newID(),
|
||||
RequestMessageID: newID(),
|
||||
Question: request.Query,
|
||||
Media: media,
|
||||
Status: "processing",
|
||||
}
|
||||
stored, created, err := s.repository.BeginTurn(ctx, turn)
|
||||
if err != nil {
|
||||
@@ -168,12 +172,18 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
|
||||
}
|
||||
turn = stored
|
||||
|
||||
displayTitle := session.BuildDisplayTitle(session.TitleIdentity{
|
||||
ConversationType: request.ConversationType,
|
||||
ExternalUserID: request.ExternalUserID,
|
||||
ExternalChatID: request.ExternalChatID,
|
||||
SenderName: request.SenderName,
|
||||
ConversationTitle: request.ConversationTitle,
|
||||
})
|
||||
mapping := core.SessionMapping{SessionKey: sessionKey, TenantKey: request.TenantKey,
|
||||
Channel: request.Channel, AgentID: agentID, ConversationType: request.ConversationType,
|
||||
ExternalSubjectID: subjectID}
|
||||
ExternalSubjectID: subjectID, DisplayTitle: displayTitle}
|
||||
sessionID, err := s.repository.ResolveSession(ctx, mapping, func(createCtx context.Context) (string, error) {
|
||||
title := fmt.Sprintf("%s %s", request.Channel, subjectID)
|
||||
return s.weknora.CreateSession(createCtx, title, "Managed by DobotAI; business state remains outside WeKnora")
|
||||
return s.weknora.CreateSession(createCtx, displayTitle, "Managed by DobotAI; business state remains outside WeKnora")
|
||||
})
|
||||
if err != nil {
|
||||
turn.Status, turn.ErrorMessage = "failed", err.Error()
|
||||
@@ -181,6 +191,9 @@ func (s *Service) ask(ctx context.Context, request core.QARequest, answerObserve
|
||||
return core.QAResult{}, fmt.Errorf("resolve weknora session: %w", err)
|
||||
}
|
||||
turn.WeKnoraSessionID = sessionID
|
||||
_ = s.repository.SyncSessionTitle(ctx, mapping, displayTitle, func(updateCtx context.Context, existingSessionID, title string) error {
|
||||
return s.weknora.UpdateSessionTitle(updateCtx, existingSessionID, title, "Managed by DobotAI; business state remains outside WeKnora")
|
||||
})
|
||||
|
||||
query := request.Query
|
||||
if request.ConversationType == core.ConversationGroup && strings.TrimSpace(request.SenderName) != "" {
|
||||
|
||||
+173
-12
@@ -13,16 +13,21 @@ import (
|
||||
)
|
||||
|
||||
type memoryRepository struct {
|
||||
mu sync.Mutex
|
||||
turns map[string]core.Turn
|
||||
sessions map[string]string
|
||||
beginTurns []core.Turn
|
||||
finishTurns []core.Turn
|
||||
deliveries []core.DeliveryUpdate
|
||||
mu sync.Mutex
|
||||
turns map[string]core.Turn
|
||||
sessions map[string]string
|
||||
sessionTitles map[string]string
|
||||
beginTurns []core.Turn
|
||||
finishTurns []core.Turn
|
||||
deliveries []core.DeliveryUpdate
|
||||
}
|
||||
|
||||
func newMemoryRepository() *memoryRepository {
|
||||
return &memoryRepository{turns: map[string]core.Turn{}, sessions: map[string]string{}}
|
||||
return &memoryRepository{
|
||||
turns: map[string]core.Turn{},
|
||||
sessions: map[string]string{},
|
||||
sessionTitles: map[string]string{},
|
||||
}
|
||||
}
|
||||
func (r *memoryRepository) BeginTurn(_ context.Context, turn core.Turn) (core.Turn, bool, error) {
|
||||
r.mu.Lock()
|
||||
@@ -52,8 +57,25 @@ func (r *memoryRepository) ResolveSession(ctx context.Context, m core.SessionMap
|
||||
return old, nil
|
||||
}
|
||||
r.sessions[m.SessionKey] = id
|
||||
r.sessionTitles[m.SessionKey] = m.DisplayTitle
|
||||
return id, nil
|
||||
}
|
||||
func (r *memoryRepository) SyncSessionTitle(ctx context.Context, m core.SessionMapping, title string, update core.SessionTitleUpdate) error {
|
||||
r.mu.Lock()
|
||||
id := r.sessions[m.SessionKey]
|
||||
current := r.sessionTitles[m.SessionKey]
|
||||
r.mu.Unlock()
|
||||
if current == title {
|
||||
return nil
|
||||
}
|
||||
if err := update(ctx, id, title); err != nil {
|
||||
return err
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.sessionTitles[m.SessionKey] = title
|
||||
r.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
func (r *memoryRepository) FinishTurn(_ context.Context, turn core.Turn) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
@@ -84,16 +106,32 @@ func (r *recordingResolver) Resolve(channel, externalChatID string) (string, err
|
||||
}
|
||||
|
||||
type fakeWeKnora struct {
|
||||
mu sync.Mutex
|
||||
sessions int
|
||||
queries []string
|
||||
requests []weknora.AgentChatRequest
|
||||
mu sync.Mutex
|
||||
sessions int
|
||||
titles []string
|
||||
updatedTitles []string
|
||||
updateErrors []error
|
||||
queries []string
|
||||
requests []weknora.AgentChatRequest
|
||||
}
|
||||
|
||||
func (f *fakeWeKnora) CreateSession(context.Context, string, string) (string, error) {
|
||||
func (f *fakeWeKnora) UpdateSessionTitle(_ context.Context, _, title, _ string) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.updatedTitles = append(f.updatedTitles, title)
|
||||
if len(f.updateErrors) == 0 {
|
||||
return nil
|
||||
}
|
||||
err := f.updateErrors[0]
|
||||
f.updateErrors = f.updateErrors[1:]
|
||||
return err
|
||||
}
|
||||
|
||||
func (f *fakeWeKnora) CreateSession(_ context.Context, title, _ string) (string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.sessions++
|
||||
f.titles = append(f.titles, title)
|
||||
return "session-1", nil
|
||||
}
|
||||
func (f *fakeWeKnora) AgentChat(_ context.Context, sessionID string, request weknora.AgentChatRequest) (weknora.ChatResult, error) {
|
||||
@@ -166,6 +204,129 @@ func TestAskUsesSharedGroupSessionAndReplaysDuplicate(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestAskCreatesReadableDirectSessionTitle(t *testing.T) {
|
||||
repo := newMemoryRepository()
|
||||
client := &fakeWeKnora{}
|
||||
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = service.Ask(context.Background(), core.QARequest{
|
||||
TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "staff-123456", ExternalMessageID: "message-direct-title",
|
||||
SenderName: " 张三 ", Query: "question",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Ask: %v", err)
|
||||
}
|
||||
client.mu.Lock()
|
||||
defer client.mu.Unlock()
|
||||
if len(client.titles) != 1 || client.titles[0] != "张三" {
|
||||
t.Fatalf("created titles = %v, want [张三]", client.titles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAskCreatesReadableGroupSessionTitle(t *testing.T) {
|
||||
repo := newMemoryRepository()
|
||||
client := &fakeWeKnora{}
|
||||
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = service.Ask(context.Background(), core.QARequest{
|
||||
TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationGroup,
|
||||
ExternalChatID: "cid-987654", ExternalUserID: "staff-1", ExternalMessageID: "message-group-title",
|
||||
SenderName: "张三", ConversationTitle: " NOVA项目群 ", Query: "question",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Ask: %v", err)
|
||||
}
|
||||
client.mu.Lock()
|
||||
defer client.mu.Unlock()
|
||||
if len(client.titles) != 1 || client.titles[0] != "NOVA项目群" {
|
||||
t.Fatalf("created titles = %v, want [NOVA项目群]", client.titles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAskSynchronizesExistingSessionTitleOnlyWhenChanged(t *testing.T) {
|
||||
repo := newMemoryRepository()
|
||||
key := "v1:tenant:dingtalk:agent-1:user:staff-1"
|
||||
repo.sessions[key] = "session-existing"
|
||||
repo.sessionTitles[key] = ""
|
||||
client := &fakeWeKnora{}
|
||||
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = service.Ask(context.Background(), core.QARequest{
|
||||
TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "staff-1", ExternalMessageID: "message-title-sync", SenderName: "张三", Query: "question",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Ask: %v", err)
|
||||
}
|
||||
client.mu.Lock()
|
||||
defer client.mu.Unlock()
|
||||
if client.sessions != 0 || len(client.updatedTitles) != 1 || client.updatedTitles[0] != "张三" {
|
||||
t.Fatalf("sessions=%d updatedTitles=%v", client.sessions, client.updatedTitles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAskSkipsExistingSessionTitleWhenUnchanged(t *testing.T) {
|
||||
repo := newMemoryRepository()
|
||||
key := "v1:tenant:dingtalk:agent-1:user:staff-1"
|
||||
repo.sessions[key] = "session-existing"
|
||||
repo.sessionTitles[key] = "张三"
|
||||
client := &fakeWeKnora{}
|
||||
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_, err = service.Ask(context.Background(), core.QARequest{
|
||||
TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "staff-1", ExternalMessageID: "message-title-unchanged", SenderName: "张三", Query: "question",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("Ask: %v", err)
|
||||
}
|
||||
client.mu.Lock()
|
||||
defer client.mu.Unlock()
|
||||
if len(client.updatedTitles) != 0 {
|
||||
t.Fatalf("unexpected title updates: %v", client.updatedTitles)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAskRetriesFailedSessionTitleUpdateWithoutBlockingQA(t *testing.T) {
|
||||
repo := newMemoryRepository()
|
||||
key := "v1:tenant:dingtalk:agent-1:user:staff-1"
|
||||
repo.sessions[key] = "session-existing"
|
||||
repo.sessionTitles[key] = ""
|
||||
client := &fakeWeKnora{updateErrors: []error{errors.New("temporary update failure"), nil}}
|
||||
service, err := New(repo, fixedResolver{}, client, session.GroupShared)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, messageID := range []string{"message-title-retry-1", "message-title-retry-2"} {
|
||||
result, askErr := service.Ask(context.Background(), core.QARequest{
|
||||
TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "staff-1", ExternalMessageID: messageID, SenderName: "张三", Query: "question",
|
||||
})
|
||||
if askErr != nil || result.Answer != "answer" {
|
||||
t.Fatalf("Ask(%s) result=%+v error=%v", messageID, result, askErr)
|
||||
}
|
||||
}
|
||||
client.mu.Lock()
|
||||
defer client.mu.Unlock()
|
||||
if client.sessions != 0 || len(client.updatedTitles) != 2 {
|
||||
t.Fatalf("sessions=%d updatedTitles=%v", client.sessions, client.updatedTitles)
|
||||
}
|
||||
repo.mu.Lock()
|
||||
defer repo.mu.Unlock()
|
||||
if repo.sessionTitles[key] != "张三" {
|
||||
t.Fatalf("tracked title=%q, want 张三", repo.sessionTitles[key])
|
||||
}
|
||||
}
|
||||
|
||||
func TestAskPropagatesImagesToWeKnora(t *testing.T) {
|
||||
repo := newMemoryRepository()
|
||||
client := &fakeWeKnora{}
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
package session
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"unicode/utf8"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
)
|
||||
|
||||
// TitleIdentity contains only the display inputs required to name a WeKnora
|
||||
// session. Stable IDs remain the session identity; titles are display-only.
|
||||
type TitleIdentity struct {
|
||||
ConversationType core.ConversationType
|
||||
ExternalUserID string
|
||||
ExternalChatID string
|
||||
SenderName string
|
||||
ConversationTitle string
|
||||
}
|
||||
|
||||
// BuildDisplayTitle returns a readable DingTalk session title without exposing
|
||||
// the full stable ID. Direct titles contain only the employee name; department
|
||||
// and region are deliberately not part of this contract.
|
||||
func BuildDisplayTitle(identity TitleIdentity) string {
|
||||
switch identity.ConversationType {
|
||||
case core.ConversationGroup:
|
||||
if title := strings.TrimSpace(identity.ConversationTitle); title != "" {
|
||||
return title
|
||||
}
|
||||
return fallbackTitle("钉钉群聊", identity.ExternalChatID)
|
||||
default:
|
||||
if name := strings.TrimSpace(identity.SenderName); name != "" {
|
||||
return name
|
||||
}
|
||||
return fallbackTitle("钉钉用户", identity.ExternalUserID)
|
||||
}
|
||||
}
|
||||
|
||||
func fallbackTitle(prefix, stableID string) string {
|
||||
stableID = strings.TrimSpace(stableID)
|
||||
if stableID == "" {
|
||||
return prefix
|
||||
}
|
||||
runes := []rune(stableID)
|
||||
if utf8.RuneCountInString(stableID) > 4 {
|
||||
runes = runes[len(runes)-4:]
|
||||
}
|
||||
return prefix + "-" + string(runes)
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package session
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
|
||||
)
|
||||
|
||||
func TestBuildDisplayTitle(t *testing.T) {
|
||||
t.Parallel()
|
||||
tests := []struct {
|
||||
name string
|
||||
in TitleIdentity
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "direct uses employee name only",
|
||||
in: TitleIdentity{
|
||||
ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "staff-123456",
|
||||
SenderName: " 张三 ",
|
||||
},
|
||||
want: "张三",
|
||||
},
|
||||
{
|
||||
name: "direct masks stable id when name unavailable",
|
||||
in: TitleIdentity{
|
||||
ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "staff-123456",
|
||||
},
|
||||
want: "钉钉用户-3456",
|
||||
},
|
||||
{
|
||||
name: "group uses conversation title",
|
||||
in: TitleIdentity{
|
||||
ConversationType: core.ConversationGroup,
|
||||
ExternalChatID: "cid-987654",
|
||||
ConversationTitle: " NOVA项目群 ",
|
||||
},
|
||||
want: "NOVA项目群",
|
||||
},
|
||||
{
|
||||
name: "group masks stable id when title unavailable",
|
||||
in: TitleIdentity{
|
||||
ConversationType: core.ConversationGroup,
|
||||
ExternalChatID: "cid-987654",
|
||||
},
|
||||
want: "钉钉群聊-7654",
|
||||
},
|
||||
{
|
||||
name: "short id remains usable",
|
||||
in: TitleIdentity{
|
||||
ConversationType: core.ConversationDirect,
|
||||
ExternalUserID: "u7",
|
||||
},
|
||||
want: "钉钉用户-u7",
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
tt := tt
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
if got := BuildDisplayTitle(tt.in); got != tt.want {
|
||||
t.Fatalf("BuildDisplayTitle() = %q, want %q", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
|
||||
"gitea.zheromiku.top/Weknora/dobotai-platform/internal/directory"
|
||||
)
|
||||
|
||||
func (s *Store) LookupDirectoryProfile(ctx context.Context, tenantKey, staffID string) (directory.Profile, bool, error) {
|
||||
tenantKey = strings.TrimSpace(tenantKey)
|
||||
staffID = strings.TrimSpace(staffID)
|
||||
if tenantKey == "" || staffID == "" {
|
||||
return directory.Profile{}, false, errors.New("tenant and staff id are required")
|
||||
}
|
||||
var profile directory.Profile
|
||||
var departmentIDs []byte
|
||||
err := s.pool.QueryRow(ctx, `SELECT staff_id,union_id,name,nickname,job_title,department_ids,department_name,main_department_name,region,employment_status,source,synced_at
|
||||
FROM directory_profiles WHERE tenant_key=$1 AND staff_id=$2`, tenantKey, staffID).Scan(
|
||||
&profile.StaffID, &profile.UnionID, &profile.Name, &profile.Nickname, &profile.JobTitle,
|
||||
&departmentIDs, &profile.DepartmentName, &profile.MainDepartmentName, &profile.Region,
|
||||
&profile.EmploymentStatus, &profile.Source, &profile.SyncedAt,
|
||||
)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return directory.Profile{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return directory.Profile{}, false, fmt.Errorf("lookup directory profile: %w", err)
|
||||
}
|
||||
if err := json.Unmarshal(departmentIDs, &profile.DepartmentIDs); err != nil {
|
||||
return directory.Profile{}, false, errors.New("decode directory department ids")
|
||||
}
|
||||
return profile, true, nil
|
||||
}
|
||||
|
||||
func (s *Store) UpsertDirectoryProfiles(ctx context.Context, tenantKey string, profiles []directory.Profile) error {
|
||||
tenantKey = strings.TrimSpace(tenantKey)
|
||||
if tenantKey == "" || len(profiles) == 0 {
|
||||
return errors.New("tenant and non-empty directory profiles are required")
|
||||
}
|
||||
for _, profile := range profiles {
|
||||
if strings.TrimSpace(profile.StaffID) == "" || strings.TrimSpace(profile.Name) == "" {
|
||||
return errors.New("directory profile requires staff id and name")
|
||||
}
|
||||
}
|
||||
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
for _, profile := range profiles {
|
||||
departmentIDs, marshalErr := json.Marshal(profile.DepartmentIDs)
|
||||
if marshalErr != nil {
|
||||
return errors.New("encode directory department ids")
|
||||
}
|
||||
syncedAt := profile.SyncedAt
|
||||
if syncedAt.IsZero() {
|
||||
syncedAt = time.Now().UTC()
|
||||
}
|
||||
_, err = tx.Exec(ctx, `INSERT INTO directory_profiles
|
||||
(tenant_key,staff_id,union_id,name,nickname,job_title,department_ids,department_name,main_department_name,region,employment_status,source,synced_at)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)
|
||||
ON CONFLICT (tenant_key,staff_id) DO UPDATE SET
|
||||
union_id=EXCLUDED.union_id,name=EXCLUDED.name,nickname=EXCLUDED.nickname,job_title=EXCLUDED.job_title,
|
||||
department_ids=EXCLUDED.department_ids,department_name=EXCLUDED.department_name,
|
||||
main_department_name=EXCLUDED.main_department_name,region=EXCLUDED.region,
|
||||
employment_status=EXCLUDED.employment_status,source=EXCLUDED.source,synced_at=EXCLUDED.synced_at,updated_at=NOW()`,
|
||||
tenantKey, profile.StaffID, profile.UnionID, profile.Name, profile.Nickname, profile.JobTitle,
|
||||
departmentIDs, profile.DepartmentName, profile.MainDepartmentName, profile.Region,
|
||||
profile.EmploymentStatus, profile.Source, syncedAt)
|
||||
if err != nil {
|
||||
return fmt.Errorf("upsert directory profile: %w", err)
|
||||
}
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestDirectoryStoreSQLContract(t *testing.T) {
|
||||
raw, err := os.ReadFile("directory.go")
|
||||
if err != nil {
|
||||
t.Fatalf("read directory.go: %v", err)
|
||||
}
|
||||
src := strings.ToUpper(string(raw))
|
||||
for _, required := range []string{
|
||||
"SELECT STAFF_ID,UNION_ID,NAME,NICKNAME,JOB_TITLE,DEPARTMENT_IDS,DEPARTMENT_NAME,MAIN_DEPARTMENT_NAME,REGION,EMPLOYMENT_STATUS,SOURCE,SYNCED_AT",
|
||||
"FROM DIRECTORY_PROFILES",
|
||||
"INSERT INTO DIRECTORY_PROFILES",
|
||||
"ON CONFLICT (TENANT_KEY,STAFF_ID) DO UPDATE",
|
||||
} {
|
||||
if !strings.Contains(src, required) {
|
||||
t.Errorf("directory.go missing SQL contract %q", required)
|
||||
}
|
||||
}
|
||||
for _, forbidden := range []string{"MOBILE", "EMAIL", "CLIENT_SECRET", "ACCESS_TOKEN", "RAW_PAYLOAD"} {
|
||||
if strings.Contains(src, forbidden) {
|
||||
t.Errorf("directory store contains forbidden sensitive field %q", forbidden)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestBeginTurnStoresDirectorySnapshotsAndGroupTitle(t *testing.T) {
|
||||
raw, err := os.ReadFile("store.go")
|
||||
if err != nil {
|
||||
t.Fatalf("read store.go: %v", err)
|
||||
}
|
||||
src := strings.ToUpper(string(raw))
|
||||
for _, required := range []string{
|
||||
"DEPARTMENT_SNAPSHOT,REGION_SNAPSHOT",
|
||||
"TURN.DEPARTMENTSNAPSHOT",
|
||||
"TURN.REGIONSNAPSHOT",
|
||||
"TURN.CONVERSATIONTITLE",
|
||||
} {
|
||||
if !strings.Contains(src, required) {
|
||||
t.Errorf("store.go missing %q", required)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestStoreSessionTitleSQLContract(t *testing.T) {
|
||||
raw, err := os.ReadFile("store.go")
|
||||
if err != nil {
|
||||
t.Fatalf("read store.go: %v", err)
|
||||
}
|
||||
src := strings.ToUpper(string(raw))
|
||||
for _, required := range []string{
|
||||
"INSERT INTO SESSION_MAPPINGS",
|
||||
"DISPLAY_TITLE",
|
||||
"SELECT WEKNORA_SESSION_ID,DISPLAY_TITLE FROM SESSION_MAPPINGS",
|
||||
"UPDATE SESSION_MAPPINGS SET DISPLAY_TITLE=",
|
||||
} {
|
||||
if !strings.Contains(src, required) {
|
||||
t.Errorf("store.go missing session-title SQL contract %q", required)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(string(raw), "func (s *Store) SyncSessionTitle") {
|
||||
t.Error("Store must implement SyncSessionTitle")
|
||||
}
|
||||
}
|
||||
@@ -67,6 +67,8 @@ func (s *Store) BeginTurn(ctx context.Context, turn core.Turn) (core.Turn, bool,
|
||||
conversationTitle := ""
|
||||
if turn.ConversationType == core.ConversationDirect {
|
||||
conversationTitle = strings.TrimSpace(turn.SenderName)
|
||||
} else if turn.ConversationType == core.ConversationGroup {
|
||||
conversationTitle = strings.TrimSpace(turn.ConversationTitle)
|
||||
}
|
||||
if err = tx.QueryRow(ctx, `INSERT INTO channel_conversations
|
||||
(id,tenant_key,channel,external_conversation_id,conversation_type,title,status)
|
||||
@@ -81,13 +83,13 @@ func (s *Store) BeginTurn(ctx context.Context, turn core.Turn) (core.Turn, bool,
|
||||
|
||||
if err = tx.QueryRow(ctx, `INSERT INTO messages
|
||||
(id,tenant_key,channel,conversation_id,actor_id,external_message_id,correlation_id,
|
||||
direction,role,message_type,text_content,media_json,status,actor_name_snapshot,safe_metadata_json)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7,'inbound','user',$8,$9,$10,'received',$11,'{}'::JSONB)
|
||||
direction,role,message_type,text_content,media_json,status,actor_name_snapshot,department_snapshot,region_snapshot,safe_metadata_json)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7,'inbound','user',$8,$9,$10,'received',$11,$12,$13,'{}'::JSONB)
|
||||
ON CONFLICT (tenant_key,channel,external_message_id) WHERE external_message_id!=''
|
||||
DO UPDATE SET updated_at=messages.updated_at
|
||||
RETURNING id`, turn.RequestMessageID, turn.TenantKey, turn.Channel, turn.ConversationID,
|
||||
turn.ActorID, turn.ExternalMessageID, turn.ID, turn.MessageType, turn.Question, mediaJSON,
|
||||
turn.SenderName).Scan(&turn.RequestMessageID); err != nil {
|
||||
turn.SenderName, turn.DepartmentSnapshot, turn.RegionSnapshot).Scan(&turn.RequestMessageID); err != nil {
|
||||
return core.Turn{}, false, err
|
||||
}
|
||||
|
||||
@@ -137,14 +139,43 @@ func (s *Store) ResolveSession(ctx context.Context, mapping core.SessionMapping,
|
||||
return "", err
|
||||
}
|
||||
_, err = tx.Exec(ctx, `INSERT INTO session_mappings
|
||||
(session_key,tenant_key,channel,agent_id,conversation_type,external_subject_id,weknora_session_id)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7)`, mapping.SessionKey, mapping.TenantKey, mapping.Channel, mapping.AgentID, string(mapping.ConversationType), mapping.ExternalSubjectID, sessionID)
|
||||
(session_key,tenant_key,channel,agent_id,conversation_type,external_subject_id,weknora_session_id,display_title)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7,$8)`, mapping.SessionKey, mapping.TenantKey, mapping.Channel, mapping.AgentID, string(mapping.ConversationType), mapping.ExternalSubjectID, sessionID, mapping.DisplayTitle)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return sessionID, tx.Commit(ctx)
|
||||
}
|
||||
|
||||
func (s *Store) SyncSessionTitle(ctx context.Context, mapping core.SessionMapping, title string, update core.SessionTitleUpdate) error {
|
||||
title = strings.TrimSpace(title)
|
||||
if strings.TrimSpace(mapping.SessionKey) == "" || title == "" || update == nil {
|
||||
return errors.New("session key, title and update callback are required")
|
||||
}
|
||||
tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, mapping.SessionKey); err != nil {
|
||||
return err
|
||||
}
|
||||
var sessionID, currentTitle string
|
||||
if err = tx.QueryRow(ctx, `SELECT weknora_session_id,display_title FROM session_mappings WHERE session_key=$1`, mapping.SessionKey).Scan(&sessionID, ¤tTitle); err != nil {
|
||||
return err
|
||||
}
|
||||
if currentTitle == title {
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
if err = update(ctx, sessionID, title); err != nil {
|
||||
return fmt.Errorf("update session title: %w", err)
|
||||
}
|
||||
if _, err = tx.Exec(ctx, `UPDATE session_mappings SET display_title=$2,updated_at=NOW() WHERE session_key=$1`, mapping.SessionKey, title); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
func (s *Store) FinishTurn(ctx context.Context, turn core.Turn) error {
|
||||
references, err := json.Marshal(turn.References)
|
||||
if err != nil {
|
||||
|
||||
@@ -96,6 +96,28 @@ func (c *Client) CreateSession(ctx context.Context, title, description string) (
|
||||
return response.Data.ID, nil
|
||||
}
|
||||
|
||||
// UpdateSessionTitle updates only the readable title/description of an
|
||||
// existing WeKnora session. The stable session ID is never changed.
|
||||
func (c *Client) UpdateSessionTitle(ctx context.Context, sessionID, title, description string) error {
|
||||
sessionID = strings.TrimSpace(sessionID)
|
||||
title = strings.TrimSpace(title)
|
||||
if sessionID == "" || title == "" {
|
||||
return errors.New("session id and title are required")
|
||||
}
|
||||
var response struct {
|
||||
Success bool `json:"success"`
|
||||
}
|
||||
if err := c.doJSON(ctx, http.MethodPut, "/api/v1/sessions/"+url.PathEscape(sessionID), map[string]string{
|
||||
"title": title, "description": description,
|
||||
}, &response); err != nil {
|
||||
return err
|
||||
}
|
||||
if !response.Success {
|
||||
return errors.New("weknora session update failed")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) AgentChat(ctx context.Context, sessionID string, request AgentChatRequest) (ChatResult, error) {
|
||||
return c.AgentChatStream(ctx, sessionID, request, nil)
|
||||
}
|
||||
@@ -153,8 +175,7 @@ func (c *Client) doJSON(ctx context.Context, method, path string, input, output
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
message, _ := io.ReadAll(io.LimitReader(resp.Body, 32<<10))
|
||||
return fmt.Errorf("weknora returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(message)))
|
||||
return fmt.Errorf("weknora returned HTTP %d", resp.StatusCode)
|
||||
}
|
||||
if err := json.NewDecoder(io.LimitReader(resp.Body, 2<<20)).Decode(output); err != nil {
|
||||
return fmt.Errorf("decode weknora response: %w", err)
|
||||
|
||||
@@ -49,6 +49,68 @@ func TestClientCreateSessionAndAgentChat(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientUpdateSessionTitle(t *testing.T) {
|
||||
var gotMethod, gotPath, gotTitle, gotDescription string
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/api/v1/sessions/session-1", func(w http.ResponseWriter, r *http.Request) {
|
||||
gotMethod = r.Method
|
||||
gotPath = r.URL.Path
|
||||
if r.Header.Get("X-API-Key") != "test-key" {
|
||||
t.Errorf("missing api key")
|
||||
}
|
||||
var body map[string]string
|
||||
if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
|
||||
t.Fatalf("decode request: %v", err)
|
||||
}
|
||||
gotTitle = body["title"]
|
||||
gotDescription = body["description"]
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_, _ = w.Write([]byte(`{"success":true,"data":{"id":"session-1"}}`))
|
||||
})
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
|
||||
client, err := NewClient(server.URL, "test-key", 5*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("NewClient: %v", err)
|
||||
}
|
||||
if err := client.UpdateSessionTitle(context.Background(), "session-1", "张三", "Managed by DobotAI"); err != nil {
|
||||
t.Fatalf("UpdateSessionTitle: %v", err)
|
||||
}
|
||||
if gotMethod != http.MethodPut || gotPath != "/api/v1/sessions/session-1" {
|
||||
t.Fatalf("request = %s %s", gotMethod, gotPath)
|
||||
}
|
||||
if gotTitle != "张三" || gotDescription != "Managed by DobotAI" {
|
||||
t.Fatalf("payload title=%q description=%q", gotTitle, gotDescription)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientUpdateSessionTitleSanitizesHTTPError(t *testing.T) {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/api/v1/sessions/session-1", func(w http.ResponseWriter, _ *http.Request) {
|
||||
http.Error(w, "raw-secret-body", http.StatusInternalServerError)
|
||||
})
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
|
||||
client, err := NewClient(server.URL, "api-secret", 5*time.Second)
|
||||
if err != nil {
|
||||
t.Fatalf("NewClient: %v", err)
|
||||
}
|
||||
err = client.UpdateSessionTitle(context.Background(), "session-1", "customer-secret-title", "description")
|
||||
if err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
for _, forbidden := range []string{"raw-secret-body", "api-secret", "customer-secret-title"} {
|
||||
if strings.Contains(err.Error(), forbidden) {
|
||||
t.Fatalf("error leaked %q: %v", forbidden, err)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(err.Error(), "HTTP 500") {
|
||||
t.Fatalf("error = %q, want HTTP status", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEFailsClosedOnPrematureEOF(t *testing.T) {
|
||||
_, err := parseSSE(strings.NewReader("data: {\"response_type\":\"answer\",\"content\":\"partial\",\"done\":false}\n\n"))
|
||||
if !errors.Is(err, ErrIncompleteStream) {
|
||||
|
||||
Reference in New Issue
Block a user