dingtalk-workspace-cli/internal/app/event_personal_oa_test.go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");

package app

import (
	"bytes"
	"context"
	"encoding/json"
	"io"
	"reflect"
	"strings"
	"testing"

	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
	"github.com/spf13/cobra"
)

func TestPersonalOAEventListAndSchemaCommands(t *testing.T) {
	list := newEventListCommand()
	list.SilenceUsage = true
	list.SilenceErrors = true
	var listOut bytes.Buffer
	list.SetOut(&listOut)
	list.SetArgs([]string{"--category", "oa"})
	if err := list.Execute(); err != nil {
		t.Fatalf("event list --category oa error = %v", err)
	}
	tests := []struct {
		eventKey   string
		properties []string
	}{
		{
			eventKey: personal.EventOAApprovalTaskCreated,
			properties: []string{
				"type", "event_id", "timestamp", "subscribe_id", "process_instance_id",
				"process_code", "task_id", "title", "status", "create_time", "event_time",
			},
		},
		{
			eventKey: personal.EventOAApprovalTaskFinished,
			properties: []string{
				"type", "event_id", "timestamp", "subscribe_id", "process_instance_id",
				"process_code", "task_id", "title", "status", "result", "create_time",
				"finish_time", "event_time",
			},
		},
		{
			eventKey: personal.EventOAApprovalTaskRedirected,
			properties: []string{
				"type", "event_id", "timestamp", "subscribe_id", "process_instance_id",
				"process_code", "task_id", "title", "status", "result", "create_time",
				"finish_time", "event_time",
			},
		},
		{
			eventKey: personal.EventOAApprovalInstanceStarted,
			properties: []string{
				"type", "event_id", "timestamp", "subscribe_id", "process_instance_id",
				"process_code", "title", "status", "create_time", "event_time",
			},
		},
		{
			eventKey: personal.EventOAApprovalInstanceCC,
			properties: []string{
				"type", "event_id", "timestamp", "subscribe_id", "process_instance_id",
				"process_code", "title", "status", "create_time", "event_time",
			},
		},
		{
			eventKey: personal.EventOAApprovalInstanceTerminated,
			properties: []string{
				"type", "event_id", "timestamp", "subscribe_id", "process_instance_id",
				"process_code", "title", "status", "create_time", "finish_time", "event_time",
			},
		},
		{
			eventKey: personal.EventOAApprovalInstanceFinished,
			properties: []string{
				"type", "event_id", "timestamp", "subscribe_id", "process_instance_id",
				"process_code", "title", "status", "result", "create_time", "finish_time",
				"event_time",
			},
		},
	}
	for _, tt := range tests {
		eventKey := tt.eventKey
		if !strings.Contains(listOut.String(), eventKey) {
			t.Fatalf("OA event list missing %s:\n%s", eventKey, listOut.String())
		}

		schema := newEventSchemaCommand()
		schema.SilenceUsage = true
		schema.SilenceErrors = true
		var schemaOut bytes.Buffer
		schema.SetOut(&schemaOut)
		schema.SetArgs([]string{eventKey, "--flatten"})
		if err := schema.Execute(); err != nil {
			t.Fatalf("event schema %s --flatten error = %v", eventKey, err)
		}
		var doc map[string]any
		if err := json.Unmarshal(schemaOut.Bytes(), &doc); err != nil {
			t.Fatalf("decode schema for %s: %v\n%s", eventKey, err, schemaOut.String())
		}
		if doc["event_key"] != eventKey || doc["rule_type"] != "all" || doc["jq_root_path"] != "." {
			t.Fatalf("schema document for %s = %#v", eventKey, doc)
		}
		schemaBody, ok := doc["schema"].(map[string]any)
		if !ok {
			t.Fatalf("schema body for %s = %#v", eventKey, doc["schema"])
		}
		properties, ok := schemaBody["properties"].(map[string]any)
		if !ok || len(properties) != len(tt.properties) {
			t.Fatalf("schema properties for %s = %#v, want %d fields", eventKey, schemaBody["properties"], len(tt.properties))
		}
		for _, name := range tt.properties {
			if _, ok := properties[name].(map[string]any); !ok {
				t.Fatalf("schema property %s for %s = %#v", name, eventKey, properties[name])
			}
		}
		if _, ok := properties["payload"]; ok {
			t.Fatalf("schema for %s exposed generic payload: %#v", eventKey, properties)
		}
	}
	if strings.Contains(listOut.String(), personal.EventMention) {
		t.Fatalf("OA category list leaked IM event:\n%s", listOut.String())
	}
}

func TestPersonalOAEventConsumeDryRunAndValidation(t *testing.T) {
	oldIdentity := personalResolveEventIdentity
	oldGet := personalGetSubscription
	t.Cleanup(func() {
		personalResolveEventIdentity = oldIdentity
		personalGetSubscription = oldGet
	})
	personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) {
		return personal.Identity{
			AccessToken:  "token",
			LocalSubject: "subject",
			ClientID:     "client",
			SourceID:     "open",
		}, nil
	}
	personalGetSubscription = func(_ *personal.Client, _ context.Context, subscribeID string) (*personal.Subscription, error) {
		switch subscribeID {
		case "oa-sub-task":
			return &personal.Subscription{
				SubscribeID: subscribeID,
				EventKey:    personal.EventOAApprovalTaskCreated,
				RuleType:    "all",
			}, nil
		case "im-sub-at":
			return &personal.Subscription{
				SubscribeID: subscribeID,
				EventKey:    personal.EventMention,
				RuleType:    "at",
			}, nil
		default:
			t.Fatalf("unexpected subscription lookup %q", subscribeID)
			return nil, nil
		}
	}

	oaEvents := []string{
		personal.EventOAApprovalTaskCreated,
		personal.EventOAApprovalTaskFinished,
		personal.EventOAApprovalTaskRedirected,
		personal.EventOAApprovalInstanceStarted,
		personal.EventOAApprovalInstanceCC,
		personal.EventOAApprovalInstanceTerminated,
		personal.EventOAApprovalInstanceFinished,
	}
	for _, eventKey := range oaEvents {
		t.Run(eventKey+"/dry-run", func(t *testing.T) {
			cmd := newEventConsumeCommand()
			cmd.SilenceUsage = true
			cmd.SilenceErrors = true
			var stderr bytes.Buffer
			cmd.SetOut(io.Discard)
			cmd.SetErr(&stderr)
			cmd.SetArgs([]string{eventKey, "--dry-run"})
			if err := cmd.Execute(); err != nil {
				t.Fatalf("OA dry-run error = %v", err)
			}
			if !strings.Contains(stderr.String(), "event_types      : "+eventKey) {
				t.Fatalf("OA dry-run does not select %s:\n%s", eventKey, stderr.String())
			}
		})

		for _, args := range [][]string{
			{"--user", "user-1"},
			{"--open-dingtalk-id", "open-user-1"},
			{"--group", "cid-1"},
			{"--query", "urgent"},
			{"--filter-json", `{"field":"content","op":"eq","value":"urgent"}`},
		} {
			name := strings.TrimPrefix(args[0], "--")
			t.Run(eventKey+"/reject-"+name, func(t *testing.T) {
				cmd := newEventConsumeCommand()
				cmd.SilenceUsage = true
				cmd.SilenceErrors = true
				cmd.SetOut(io.Discard)
				cmd.SetErr(io.Discard)
				cmd.SetArgs(append([]string{eventKey}, append(args, "--dry-run")...))
				err := cmd.Execute()
				if err == nil || !strings.Contains(err.Error(), "not supported") {
					t.Fatalf("OA consume %s error = %v, want unsupported option", args[0], err)
				}
			})
		}
	}

	t.Run("multi-dry-run", func(t *testing.T) {
		cmd := newEventConsumeCommand()
		cmd.SilenceUsage = true
		cmd.SilenceErrors = true
		var stderr bytes.Buffer
		cmd.SetOut(io.Discard)
		cmd.SetErr(&stderr)
		cmd.SetArgs(append(append([]string(nil), oaEvents...), "--dry-run"))
		if err := cmd.Execute(); err != nil {
			t.Fatalf("multi OA dry-run error = %v", err)
		}
		for _, eventKey := range oaEvents {
			want := "event_key=" + eventKey + " rule_type=all rule_param={}"
			if !strings.Contains(stderr.String(), want) {
				t.Fatalf("multi OA dry-run missing %q:\n%s", want, stderr.String())
			}
		}
	})

	reuseOverrides := [][]string{
		{"--user", "user-1"},
		{"--open-dingtalk-id", "open-user-1"},
		{"--group", "cid-1"},
		{"--query", "urgent"},
		{"--filter-json", `{"field":"content","op":"eq","value":"urgent"}`},
	}
	t.Run("reuse-dry-run/implicit-event-key/resolves-oa-event", func(t *testing.T) {
		cmd := newEventConsumeCommand()
		cmd.SilenceUsage = true
		cmd.SilenceErrors = true
		var stderr bytes.Buffer
		cmd.SetOut(io.Discard)
		cmd.SetErr(&stderr)
		cmd.SetArgs([]string{"--subscribe-id", "oa-sub-task", "--dry-run"})
		if err := cmd.Execute(); err != nil {
			t.Fatalf("implicit reused OA dry-run error = %v", err)
		}
		if !strings.Contains(stderr.String(), "event_types      : "+personal.EventOAApprovalTaskCreated) {
			t.Fatalf("implicit reused OA dry-run did not resolve event key:\n%s", stderr.String())
		}
	})
	for _, explicitEventKey := range []bool{true, false} {
		mode := "implicit-event-key"
		if explicitEventKey {
			mode = "explicit-event-key"
		}
		for _, override := range reuseOverrides {
			flag := override[0]
			t.Run("reuse-dry-run/"+mode+"/"+strings.TrimPrefix(flag, "--"), func(t *testing.T) {
				args := make([]string, 0, 6)
				if explicitEventKey {
					args = append(args, personal.EventOAApprovalTaskCreated)
				}
				args = append(args, "--subscribe-id", "oa-sub-task", flag, override[1], "--dry-run")
				cmd := newEventConsumeCommand()
				cmd.SilenceUsage = true
				cmd.SilenceErrors = true
				cmd.SetOut(io.Discard)
				cmd.SetErr(io.Discard)
				cmd.SetArgs(args)
				err := cmd.Execute()
				if err == nil || !strings.Contains(err.Error(), flag+" not supported for OA event") {
					t.Fatalf("%s reused OA dry-run %s error = %v", mode, flag, err)
				}
			})
		}
	}

	t.Run("reuse-dry-run/implicit-im-remains-supported", func(t *testing.T) {
		cmd := newEventConsumeCommand()
		cmd.SilenceUsage = true
		cmd.SilenceErrors = true
		cmd.SetOut(io.Discard)
		cmd.SetErr(io.Discard)
		cmd.SetArgs([]string{"--subscribe-id", "im-sub-at", "--query", "urgent", "--dry-run"})
		if err := cmd.Execute(); err != nil {
			t.Fatalf("implicit reused IM dry-run error = %v", err)
		}
	})

	for _, override := range reuseOverrides {
		flag, value := override[0], override[1]
		t.Run("multi-reject-"+strings.TrimPrefix(flag, "--"), func(t *testing.T) {
			cmd := newEventConsumeCommand()
			cmd.SilenceUsage = true
			cmd.SilenceErrors = true
			cmd.SetOut(io.Discard)
			cmd.SetErr(io.Discard)
			args := append([]string(nil), oaEvents...)
			args = append(args, flag, value, "--dry-run")
			cmd.SetArgs(args)
			err := cmd.Execute()
			if err == nil || !strings.Contains(err.Error(), "not supported for OA event") {
				t.Fatalf("multi OA consume %s error = %v", flag, err)
			}
		})
	}

	for _, test := range []struct {
		name string
		args []string
	}{
		{
			name: "message query remains supported",
			args: []string{personal.EventMention, "--query", "urgent", "--dry-run"},
		},
		{
			name: "single group lifecycle filter remains supported",
			args: []string{
				personal.EventGroupUpdated,
				"--group", "cid-1",
				"--filter-json", `{"field":"future","op":"eq","value":"value"}`,
				"--dry-run",
			},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			cmd := newEventConsumeCommand()
			cmd.SilenceUsage = true
			cmd.SilenceErrors = true
			cmd.SetOut(io.Discard)
			cmd.SetErr(io.Discard)
			cmd.SetArgs(test.args)
			if err := cmd.Execute(); err != nil {
				t.Fatalf("existing IM consume behavior changed: %v", err)
			}
		})
	}
}

func TestCrossPlatformCoveragePersonalOAValidationBranches(t *testing.T) {
	invalid := personalConsumeOptions{
		EventKey: personal.EventOAApprovalTaskCreated,
		UserID:   "user-1",
	}
	if err := validatePersonalSubscriptionOptions(invalid); err == nil ||
		!strings.Contains(err.Error(), "--user not supported for OA event") {
		t.Fatalf("validatePersonalSubscriptionOptions() error = %v", err)
	}
	if _, err := preparePersonalSubscription(personal.Identity{}, invalid); err == nil ||
		!strings.Contains(err.Error(), "--user not supported for OA event") {
		t.Fatalf("preparePersonalSubscription() error = %v", err)
	}

	oldGet := personalGetSubscription
	t.Cleanup(func() { personalGetSubscription = oldGet })
	personalGetSubscription = func(*personal.Client, context.Context, string) (*personal.Subscription, error) {
		return &personal.Subscription{
			SubscribeID: "oa-sub-without-event-key",
			RuleType:    "all",
		}, nil
	}
	_, _, _, err := ensurePersonalSubscription(
		context.Background(),
		nil,
		personal.Identity{},
		personalConsumeOptions{
			SubscribeID: "oa-sub-without-event-key",
			EventKey:    personal.EventOAApprovalTaskCreated,
			UserID:      "user-1",
		},
	)
	if err == nil || !strings.Contains(err.Error(), "--user not supported for OA event") {
		t.Fatalf("ensurePersonalSubscription() error = %v", err)
	}
}

func TestPersonalOAMultiConsumeCreatesIndependentAllSubscriptionsOnSharedBus(t *testing.T) {
	restoreMany := installPersonalManySeams(t)
	defer restoreMany()
	oldCreate := personalCreateSubscription
	defer func() { personalCreateSubscription = oldCreate }()
	t.Setenv("DWS_CONFIG_DIR", t.TempDir())

	identity := personal.Identity{
		AccessToken:  "token",
		LocalSubject: "subject",
		ClientID:     "client",
		SourceID:     "open",
	}
	personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) {
		return identity, nil
	}
	var requests []personal.CreateSubscriptionRequest
	personalCreateSubscription = func(_ *personal.Client, _ context.Context, req personal.CreateSubscriptionRequest) (*personal.Subscription, error) {
		requests = append(requests, req)
		return &personal.Subscription{SubscribeID: "sub-" + req.EventKey}, nil
	}
	personalEnsureSubscription = ensurePersonalSubscription
	var states []personal.RunState
	personalUpsertRunState = func(_ string, state personal.RunState) error {
		states = append(states, state)
		return nil
	}
	personalDeleteSubscription = func(*personal.Client, context.Context, string) error { return nil }
	personalRemoveRunStates = func(string, []string) error { return nil }
	personalValidateConsumeConfig = func(consume.Config) error { return nil }
	runManyCalls := 0
	var gotSpecs []consume.ConsumerSpec
	personalConsumeRunMany = func(_ context.Context, _ consume.Config, specs []consume.ConsumerSpec) error {
		runManyCalls++
		gotSpecs = append([]consume.ConsumerSpec(nil), specs...)
		return nil
	}

	eventKeys := []string{
		personal.EventOAApprovalTaskCreated,
		personal.EventOAApprovalTaskFinished,
		personal.EventOAApprovalTaskRedirected,
		personal.EventOAApprovalInstanceStarted,
		personal.EventOAApprovalInstanceCC,
		personal.EventOAApprovalInstanceTerminated,
		personal.EventOAApprovalInstanceFinished,
	}
	if err := runPersonalEventConsume(newPersonalCoverageCommand(), personalConsumeOptions{
		EventKeys: eventKeys,
		Flatten:   true,
	}); err != nil {
		t.Fatalf("multi OA consume error = %v", err)
	}
	if runManyCalls != 1 {
		t.Fatalf("RunMany calls = %d, want one shared-bus consume call", runManyCalls)
	}
	if len(requests) != len(eventKeys) || len(states) != len(eventKeys) || len(gotSpecs) != len(eventKeys) {
		t.Fatalf("requests=%d states=%d specs=%d, want %d each", len(requests), len(states), len(gotSpecs), len(eventKeys))
	}
	for i, eventKey := range eventKeys {
		req := requests[i]
		if req.EventKey != eventKey || req.RuleType != "all" || req.RuleParam == nil || len(req.RuleParam) != 0 || req.Filter != nil {
			t.Fatalf("subscription request[%d] = %#v, want %s all/{}", i, req, eventKey)
		}
		if states[i].EventKey != eventKey || states[i].RuleType != "all" {
			t.Fatalf("run state[%d] = %#v", i, states[i])
		}
		wantSpec := consume.ConsumerSpec{
			EventKey:         eventKey,
			EventTypes:       []string{eventKey},
			SubscribeID:      "sub-" + eventKey,
			ReadySubscribeID: "sub-" + eventKey,
		}
		if !reflect.DeepEqual(gotSpecs[i], wantSpec) {
			t.Fatalf("consumer spec[%d] = %#v, want %#v", i, gotSpecs[i], wantSpec)
		}
	}
}

func TestPersonalOAReusedSubscriptionRejectsDefinitionOverridesAtRuntime(t *testing.T) {
	oldGet := personalGetSubscription
	t.Cleanup(func() { personalGetSubscription = oldGet })
	getCalls := 0
	personalGetSubscription = func(*personal.Client, context.Context, string) (*personal.Subscription, error) {
		getCalls++
		return &personal.Subscription{
			SubscribeID: "oa-sub-task",
			EventKey:    personal.EventOAApprovalTaskCreated,
			RuleType:    "all",
		}, nil
	}

	tests := []struct {
		name string
		set  func(*personalConsumeOptions)
	}{
		{name: "user", set: func(opts *personalConsumeOptions) { opts.UserID = "user-1" }},
		{name: "open-dingtalk-id", set: func(opts *personalConsumeOptions) { opts.OpenDingTalkID = "open-user-1" }},
		{name: "group", set: func(opts *personalConsumeOptions) { opts.GroupID = "cid-1" }},
		{name: "query", set: func(opts *personalConsumeOptions) { opts.QueryCSV = "urgent" }},
		{name: "filter-json", set: func(opts *personalConsumeOptions) { opts.FilterJSON = `{"field":"content","op":"eq","value":"urgent"}` }},
	}
	for _, explicitEventKey := range []bool{true, false} {
		mode := "implicit-event-key"
		if explicitEventKey {
			mode = "explicit-event-key"
		}
		for _, test := range tests {
			t.Run(mode+"/"+test.name, func(t *testing.T) {
				opts := personalConsumeOptions{SubscribeID: "oa-sub-task"}
				if explicitEventKey {
					opts.EventKey = personal.EventOAApprovalTaskCreated
				}
				test.set(&opts)
				before := getCalls
				_, _, _, err := ensurePersonalSubscription(
					context.Background(),
					nil,
					personal.Identity{},
					opts,
				)
				if err == nil || !strings.Contains(err.Error(), "--"+test.name+" not supported for OA event") {
					t.Fatalf("reused OA subscription %s error = %v", test.name, err)
				}
				if getCalls != before+1 {
					t.Fatalf("subscription lookup calls = %d, want %d", getCalls, before+1)
				}
			})
		}
	}
}

func TestPersonalOAImplicitReuseRuntimeLooksUpEventBeforeValidation(t *testing.T) {
	oldIdentity := personalResolveEventIdentity
	oldGet := personalGetSubscription
	t.Cleanup(func() {
		personalResolveEventIdentity = oldIdentity
		personalGetSubscription = oldGet
	})
	personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) {
		return personal.Identity{
			AccessToken:  "token",
			LocalSubject: "subject",
			ClientID:     "client",
			SourceID:     "open",
		}, nil
	}
	getCalls := 0
	personalGetSubscription = func(*personal.Client, context.Context, string) (*personal.Subscription, error) {
		getCalls++
		return &personal.Subscription{
			SubscribeID: "oa-sub-task",
			EventKey:    personal.EventOAApprovalTaskCreated,
			RuleType:    "all",
		}, nil
	}

	cmd := newEventConsumeCommand()
	cmd.SilenceUsage = true
	cmd.SilenceErrors = true
	cmd.SetOut(io.Discard)
	cmd.SetErr(io.Discard)
	cmd.SetArgs([]string{"--subscribe-id", "oa-sub-task", "--group", "cid-1"})
	err := cmd.Execute()
	if err == nil || !strings.Contains(err.Error(), "--group not supported for OA event "+personal.EventOAApprovalTaskCreated) {
		t.Fatalf("implicit reused OA runtime error = %v", err)
	}
	if getCalls != 1 {
		t.Fatalf("subscription lookup calls = %d, want 1", getCalls)
	}
}

func TestPersonalIMReusedSubscriptionWithExistingOverridesRemainsSupported(t *testing.T) {
	oldGet := personalGetSubscription
	t.Cleanup(func() { personalGetSubscription = oldGet })
	personalGetSubscription = func(*personal.Client, context.Context, string) (*personal.Subscription, error) {
		return &personal.Subscription{
			SubscribeID: "im-sub",
			EventKey:    personal.EventSingleChat,
			RuleType:    "singleChat",
		}, nil
	}

	sub, eventKey, ruleType, err := ensurePersonalSubscription(
		context.Background(),
		nil,
		personal.Identity{},
		personalConsumeOptions{
			SubscribeID: "im-sub",
			EventKey:    personal.EventSingleChat,
			UserID:      "user-1",
			QueryCSV:    "urgent",
			FilterJSON:  `{"field":"content","op":"eq","value":"urgent"}`,
		},
	)
	if err != nil {
		t.Fatalf("reused IM subscription error = %v", err)
	}
	if sub.SubscribeID != "im-sub" || eventKey != personal.EventSingleChat || ruleType != "singleChat" {
		t.Fatalf("reused IM subscription = %#v, event=%q rule=%q", sub, eventKey, ruleType)
	}
}

func TestPersonalOAStatusAndStopCommandWiring(t *testing.T) {
	oldStatus := eventRunPersonalStatus
	oldStop := eventRunPersonalStop
	t.Cleanup(func() {
		eventRunPersonalStatus = oldStatus
		eventRunPersonalStop = oldStop
	})

	var statusOpts personalStatusOptions
	eventRunPersonalStatus = func(_ *cobra.Command, opts personalStatusOptions) error {
		statusOpts = opts
		return nil
	}
	status := newEventStatusCommand()
	status.SilenceUsage = true
	status.SilenceErrors = true
	status.SetOut(io.Discard)
	status.SetErr(io.Discard)
	status.SetArgs([]string{
		"--event", personal.EventOAApprovalTaskCreated,
		"--subscribe-id", "oa-sub-task",
		"--status", "all",
	})
	if err := status.Execute(); err != nil {
		t.Fatalf("OA event status error = %v", err)
	}
	if statusOpts.EventKey != personal.EventOAApprovalTaskCreated ||
		statusOpts.SubscribeID != "oa-sub-task" ||
		statusOpts.Status != "all" {
		t.Fatalf("OA status options = %#v", statusOpts)
	}

	var stopOpts personalStopOptions
	eventRunPersonalStop = func(_ *cobra.Command, opts personalStopOptions) error {
		stopOpts = opts
		return nil
	}
	root := &cobra.Command{Use: "dws", SilenceUsage: true, SilenceErrors: true}
	root.SetOut(io.Discard)
	root.SetErr(io.Discard)
	root.PersistentFlags().Bool("yes", false, "")
	event := &cobra.Command{Use: "event"}
	event.AddCommand(newEventStopCommand())
	root.AddCommand(event)
	root.SetArgs([]string{"event", "stop", "oa-sub-task", "--yes"})
	if err := root.Execute(); err != nil {
		t.Fatalf("OA event stop error = %v", err)
	}
	if stopOpts.SubscribeID != "oa-sub-task" || stopOpts.All {
		t.Fatalf("OA stop options = %#v", stopOpts)
	}
}