Files
aa6abc5ed6 feat(event): add personal event subscriptions (#589)
* dws event

* fix consume fail

* test: add stream ticket injection probe

* feat: add portal ticket stream mode

* user event

* fix: allow portal ticket normal without app secret

* event

* user event

* eventType filter

* refactor(event): 优化IPC端点路径处理和改进相关测试

- 用dwsevent.IPCEndpoint替代原先根据GOOS判断的路径逻辑
- 新增event包实现Unix socket路径长度限制及长路径fallback机制
- 添加endpoint_test.go覆盖路径短长及唯一性的单元测试
- 修改busctl模块使用统一的IPC端点获取方法,避免重复实现
- transport_unix.go新增checkSocketPath函数检查路径长度,防止EINVAL错误
- 在监听和连接Unix socket时加入路径限制检查,提升错误明晰度
- 去除多个文件中无用的runtime导入,简化代码依赖

* opt

* event skill

* default value

* install script event

* fix: remove subscribe id event fanout filter

* fix(personal): 修正指定发送人消息描述错误

* more im event

* filter subId

* fix: align personal event schema with stream payload

* fix: avoid duplicate app helper name

* feat: simplify personal event schemas

* feat: simplify event schema output

* docs: refine dingtalk event skill references

* feat: align personal event consume flags

* fix event stop and status visibility

* hide app event public entrypoints

* hide incomplete personal sender event

* remove external event reference comments

* chore(event): prepare official release

* fix(event): harden personal stream lifecycle

---------

Co-authored-by: 玉澜 <yulan.wqy@alibaba-inc.com>
Co-authored-by: zhengyubai <zhengyubai618@gmail.com>
2026-07-10 17:54:43 +08:00

236 lines
6.7 KiB
Go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package source
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
"github.com/open-dingtalk/dingtalk-stream-sdk-go/event"
"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
)
func TestNew_RequiresClientID(t *testing.T) {
if _, err := New(Config{ClientSecret: "secret"}); err == nil {
t.Fatal("expected error when ClientID empty")
}
}
func TestNew_RequiresClientSecret(t *testing.T) {
if _, err := New(Config{ClientID: "id"}); err == nil {
t.Fatal("expected error when ClientSecret empty")
}
}
func TestNew_DefaultsNow(t *testing.T) {
s, err := New(Config{ClientID: "id", ClientSecret: "secret"})
if err != nil {
t.Fatalf("New: %v", err)
}
if s.cfg.Now == nil {
t.Fatal("Now should default to time.Now")
}
}
func TestNew_AcceptsPortalTicketNormalWithoutClientSecret(t *testing.T) {
if _, err := New(Config{
PortalTicket: &PortalTicketConfig{
TicketURL: "https://example.com/stream/connections/ticket",
AccessToken: "token",
SourceID: "pre_open_source",
Mode: "normal",
},
}); err != nil {
t.Fatalf("New: %v", err)
}
}
func TestNew_RejectsPortalTicketCustomWithoutSecret(t *testing.T) {
if _, err := New(Config{
ClientID: "id",
PortalTicket: &PortalTicketConfig{
TicketURL: "https://example.com/stream/connections/ticket",
AccessToken: "token",
SourceID: "pre_open_source",
Mode: "custom",
ClientID: "custom_client",
},
}); err == nil {
t.Fatal("expected error when custom portal ticket secret is empty")
}
}
func TestRequestPortalTicketCustomBody(t *testing.T) {
var got struct {
SourceID string `json:"sourceId"`
ChannelType string `json:"channelType"`
Mode string `json:"mode"`
ClientID string `json:"clientId"`
ClientSecret string `json:"clientSecret"`
}
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if token := r.Header.Get("x-user-access-token"); token != "token-123" {
t.Fatalf("x-user-access-token = %q", token)
}
if err := json.NewDecoder(r.Body).Decode(&got); err != nil {
t.Fatalf("decode request: %v", err)
}
_ = json.NewEncoder(w).Encode(map[string]any{
"success": true,
"result": map[string]string{
"endpoint": "wss://pre-wss-open-connection.dingtalk.com/connect",
"ticket": "ticket-123",
},
})
}))
defer srv.Close()
ticket, err := requestPortalTicket(context.Background(), &PortalTicketConfig{
TicketURL: srv.URL,
AccessToken: "token-123",
SourceID: "pre_open_source",
Mode: "custom",
ClientID: "custom_client",
ClientSecret: "custom_secret",
})
if err != nil {
t.Fatalf("requestPortalTicket: %v", err)
}
if ticket.Endpoint == "" || ticket.Ticket == "" {
t.Fatalf("ticket = %#v", ticket)
}
if got.SourceID != "pre_open_source" || got.ChannelType != "pre_open_source" {
t.Fatalf("source fields = %#v", got)
}
if got.Mode != "custom" || got.ClientID != "custom_client" || got.ClientSecret != "custom_secret" {
t.Fatalf("custom fields = %#v", got)
}
}
func TestState_InitiallyDisconnected(t *testing.T) {
s, _ := New(Config{ClientID: "id", ClientSecret: "secret"})
if got := s.State().State; got != StateDisconnected {
t.Fatalf("initial state = %s, want %s", got, StateDisconnected)
}
}
func TestMakeHandler_TranslatesDataFrameToRawEvent(t *testing.T) {
fixed := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC)
s, err := New(Config{
ClientID: "id",
ClientSecret: "secret",
Now: func() time.Time { return fixed },
})
if err != nil {
t.Fatalf("New: %v", err)
}
var got *dwsevent.RawEvent
handler := s.makeHandler(func(e *dwsevent.RawEvent) { got = e })
df := &payload.DataFrame{
SpecVersion: "1.0",
Type: "EVENT",
Time: 1700000000,
Headers: payload.DataFrameHeader{
"eventId": "ev_abc",
"eventBornTime": "1700000000123",
"eventCorpId": "corp_x",
"eventType": "im.message.receive_v1",
"eventUnifiedAppId": "app_y",
"extra": "passthrough",
},
Data: `{"chat":"hello"}`,
}
resp, err := handler(context.Background(), df)
if err != nil {
t.Fatalf("handler: %v", err)
}
if resp == nil {
t.Fatal("handler returned nil resp")
}
if got == nil {
t.Fatal("emit was not called")
}
if got.EventID != "ev_abc" {
t.Errorf("EventID = %q, want %q", got.EventID, "ev_abc")
}
if got.EventType != "im.message.receive_v1" {
t.Errorf("EventType = %q", got.EventType)
}
if got.EventCorpID != "corp_x" {
t.Errorf("EventCorpID = %q", got.EventCorpID)
}
if got.EventUnifiedAppID != "app_y" {
t.Errorf("EventUnifiedAppID = %q", got.EventUnifiedAppID)
}
if got.EventBornTime != 1700000000123 {
t.Errorf("EventBornTime = %d", got.EventBornTime)
}
if got.Data != `{"chat":"hello"}` {
t.Errorf("Data = %q", got.Data)
}
if !got.ReceivedAt.Equal(fixed) {
t.Errorf("ReceivedAt = %v, want %v (injected Now)", got.ReceivedAt, fixed)
}
if got.Headers["extra"] != "passthrough" {
t.Error("passthrough header lost")
}
// Also verify the connection state machine ticked: handler should have
// called OnEvent, but since we never called OnConnected the visible
// state remains disconnected. We only verify the lastEventAt timestamp.
snap := s.State()
if snap.LastEventAt.IsZero() {
t.Error("OnEvent should have updated lastEventAt")
}
}
func TestStart_RejectsNilEmit(t *testing.T) {
s, _ := New(Config{ClientID: "id", ClientSecret: "secret"})
err := s.Start(context.Background(), nil)
if err == nil || err.Error() == "" {
t.Fatalf("Start with nil emit should error, got %v", err)
}
}
func TestMakeHandler_NilHeaders(t *testing.T) {
s, _ := New(Config{ClientID: "id", ClientSecret: "secret"})
handler := s.makeHandler(func(*dwsevent.RawEvent) {})
df := &payload.DataFrame{
// no Headers at all
Data: "{}",
}
resp, err := handler(context.Background(), df)
if err != nil {
t.Fatalf("handler: %v", err)
}
if resp == nil {
t.Fatal("nil resp")
}
}
// Compile-time guard: the EventProcessResult we return must be the SDK's
// EventProcessResultSuccess type. If the SDK ever changes the shape, this
// will fail to compile.
var _ = event.NewEventProcessResultSuccess