feat(pico): add POST /pico/message HTTP ingress endpoint
Add an authenticated HTTP ingress path so external systems can trigger agent processing without a persistent WebSocket connection. - Add POST /pico/message handler with auth, body-size limiting, JSON parsing, and bus publish - Harden token checks with crypto/subtle.ConstantTimeCompare - Add comprehensive unit tests for routing, auth variants, request validation, and published bus semantics - Update channel docs (EN/ZH)
This commit is contained in:
Linus Brossberg
parent
bb2eddc79d
commit
968799bbde
4 changed files with 495 additions and 4 deletions
|
|
@ -1372,7 +1372,7 @@ agentLoop.Stop() // Stop Agent
|
|||
|
||||
3. **WeCom is now a single channel**: `"wecom"` is implemented as a WebSocket-based AI Bot channel with route persistence. Access control uses the shared channel allowlist mechanism. It no longer exposes the legacy webhook/app split.
|
||||
|
||||
4. **Pico Protocol**: `pkg/channels/pico/` implements a custom PicoClaw native protocol channel that receives messages via WebSocket webhook (`/pico/ws`).
|
||||
4. **Pico Protocol**: `pkg/channels/pico/` implements a custom PicoClaw native protocol channel that receives messages via WebSocket (`/pico/ws`) and HTTP message ingress (`POST /pico/message`). The HTTP endpoint accepts a JSON body with `content` and optional `session_id`, authenticates with the same Pico token as the WebSocket (Bearer header, or query parameter when `allow_token_query` is enabled), and publishes the message to the bus asynchronously (returns `202 Accepted`).
|
||||
|
||||
5. **WhatsApp has two modes**: `"whatsapp"` (Bridge mode, communicates via external bridge URL) and `"whatsapp_native"` (native whatsmeow mode, connects directly to WhatsApp). Manager selects which to initialize based on `WhatsAppConfig.UseNative`.
|
||||
|
||||
|
|
|
|||
|
|
@ -1371,7 +1371,7 @@ agentLoop.Stop() // 停止 Agent
|
|||
|
||||
3. **WeCom 现在只有一个 channel**：`"wecom"` 采用 WebSocket AI Bot 实现，带路由持久化；访问控制走统一的 channel 白名单机制，不再保留旧的 webhook/app 双分支。
|
||||
|
||||
4. **Pico Protocol**：`pkg/channels/pico/` 实现了一个自定义的 PicoClaw 原生协议 channel，通过 WebSocket webhook (`/pico/ws`) 接收消息。
|
||||
4. **Pico Protocol**：`pkg/channels/pico/` 实现了一个自定义的 PicoClaw 原生协议 channel，通过 WebSocket (`/pico/ws`) 和 HTTP 消息入口 (`POST /pico/message`) 接收消息。HTTP 端点接受包含 `content` 和可选 `session_id` 的 JSON 请求体，使用与 WebSocket 相同的 Pico token 进行认证（Bearer header，或在启用 `allow_token_query` 时支持查询参数），并将消息异步发布到总线（返回 `202 Accepted`）。
|
||||
|
||||
5. **WhatsApp 有两种模式**：`"whatsapp"`（Bridge 模式，通过外部 bridge URL 通信）和 `"whatsapp_native"`（原生 whatsmeow 模式，直接连接 WhatsApp）。Manager 根据 `WhatsAppConfig.UseNative` 决定初始化哪个。
|
||||
|
||||
|
|
|
|||
|
|
@ -2,8 +2,10 @@ package pico
|
|||
|
||||
import (
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync"
|
||||
|
|
@ -20,6 +22,23 @@ import (
|
|||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
)
|
||||
|
||||
// maxHTTPBodySize is the maximum request body for HTTP message ingress.
|
||||
// Text prompts are small; 64 KiB is generous while keeping the attack surface low.
|
||||
const maxHTTPBodySize = 64 << 10 // 64 KiB
|
||||
|
||||
// httpMessageRequest is the JSON body for POST /pico/message.
|
||||
type httpMessageRequest struct {
|
||||
Content string `json:"content"`
|
||||
SessionID string `json:"session_id,omitempty"`
|
||||
}
|
||||
|
||||
// httpMessageResponse is the JSON response for POST /pico/message.
|
||||
type httpMessageResponse struct {
|
||||
OK bool `json:"ok"`
|
||||
Status string `json:"status"`
|
||||
SessionID string `json:"session_id"`
|
||||
}
|
||||
|
||||
// picoConn represents a single WebSocket connection.
|
||||
type picoConn struct {
|
||||
id string
|
||||
|
|
@ -228,6 +247,8 @@ func (c *PicoChannel) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|||
switch path {
|
||||
case "/ws", "/ws/":
|
||||
c.handleWebSocket(w, r)
|
||||
case "/message", "/message/":
|
||||
c.handleHTTPMessage(w, r)
|
||||
default:
|
||||
http.NotFound(w, r)
|
||||
}
|
||||
|
|
@ -383,6 +404,8 @@ func (c *PicoChannel) handleWebSocket(w http.ResponseWriter, r *http.Request) {
|
|||
// 1. Authorization: Bearer <token> header
|
||||
// 2. Sec-WebSocket-Protocol "token.<value>" (for browsers that can't set headers)
|
||||
// 3. Query parameter "token" (only when AllowTokenQuery is on)
|
||||
//
|
||||
// Token comparison uses subtle.ConstantTimeCompare to reduce timing side-channels.
|
||||
func (c *PicoChannel) authenticate(r *http.Request) bool {
|
||||
token := c.config.Token()
|
||||
if token == "" {
|
||||
|
|
@ -392,7 +415,7 @@ func (c *PicoChannel) authenticate(r *http.Request) bool {
|
|||
// Check Authorization header
|
||||
auth := r.Header.Get("Authorization")
|
||||
if after, ok := strings.CutPrefix(auth, "Bearer "); ok {
|
||||
if after == token {
|
||||
if subtle.ConstantTimeCompare([]byte(after), []byte(token)) == 1 {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
|
@ -404,7 +427,7 @@ func (c *PicoChannel) authenticate(r *http.Request) bool {
|
|||
|
||||
// Check query parameter only when explicitly allowed
|
||||
if c.config.AllowTokenQuery {
|
||||
if r.URL.Query().Get("token") == token {
|
||||
if subtle.ConstantTimeCompare([]byte(r.URL.Query().Get("token")), []byte(token)) == 1 {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
|
@ -424,6 +447,89 @@ func (c *PicoChannel) matchedSubprotocol(r *http.Request) string {
|
|||
return ""
|
||||
}
|
||||
|
||||
// handleHTTPMessage processes an inbound HTTP message trigger.
|
||||
// This is the HTTP equivalent of a Pico WebSocket message.send — it accepts
|
||||
// a JSON body, publishes the message to the bus, and returns 202 Accepted.
|
||||
func (c *PicoChannel) handleHTTPMessage(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
if !c.IsRunning() {
|
||||
http.Error(w, "channel not running", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
|
||||
if !c.authenticate(r) {
|
||||
http.Error(w, "unauthorized", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
|
||||
body, err := io.ReadAll(io.LimitReader(r.Body, maxHTTPBodySize+1))
|
||||
if err != nil {
|
||||
http.Error(w, "bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if int64(len(body)) > maxHTTPBodySize {
|
||||
http.Error(w, "request entity too large", http.StatusRequestEntityTooLarge)
|
||||
return
|
||||
}
|
||||
|
||||
var req httpMessageRequest
|
||||
if err := json.Unmarshal(body, &req); err != nil {
|
||||
http.Error(w, "bad request: invalid JSON", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
// Validate non-empty, but preserve original content including whitespace.
|
||||
// This matches WebSocket message.send semantics in handleMessageSend.
|
||||
content := req.Content
|
||||
if strings.TrimSpace(content) == "" {
|
||||
http.Error(w, "bad request: empty content", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
sessionID := req.SessionID
|
||||
if sessionID == "" {
|
||||
sessionID = uuid.New().String()
|
||||
}
|
||||
|
||||
chatID := "pico:" + sessionID
|
||||
senderID := "pico-user"
|
||||
peer := bus.Peer{Kind: "direct", ID: "pico:" + sessionID}
|
||||
|
||||
metadata := map[string]string{
|
||||
"platform": "pico",
|
||||
"session_id": sessionID,
|
||||
"transport": "http",
|
||||
}
|
||||
|
||||
logger.DebugCF("pico", "Received HTTP message", map[string]any{
|
||||
"session_id": sessionID,
|
||||
"preview": truncate(content, 50),
|
||||
})
|
||||
|
||||
sender := bus.SenderInfo{
|
||||
Platform: "pico",
|
||||
PlatformID: senderID,
|
||||
CanonicalID: identity.BuildCanonicalID("pico", senderID),
|
||||
}
|
||||
|
||||
// Allow-list enforcement is handled internally by BaseChannel.HandleMessage,
|
||||
// matching WebSocket message.send semantics (silent drop, not HTTP error).
|
||||
msgID := uuid.New().String()
|
||||
c.HandleMessage(c.ctx, peer, msgID, senderID, chatID, content, nil, metadata, sender)
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusAccepted)
|
||||
json.NewEncoder(w).Encode(httpMessageResponse{
|
||||
OK: true,
|
||||
Status: "accepted",
|
||||
SessionID: sessionID,
|
||||
})
|
||||
}
|
||||
|
||||
// readLoop reads messages from a WebSocket connection.
|
||||
func (c *PicoChannel) readLoop(pc *picoConn) {
|
||||
defer func() {
|
||||
|
|
|
|||
|
|
@ -1,17 +1,25 @@
|
|||
package pico
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/bus"
|
||||
"github.com/sipeed/picoclaw/pkg/channels"
|
||||
"github.com/sipeed/picoclaw/pkg/config"
|
||||
)
|
||||
|
||||
const testToken = "test-pico-token"
|
||||
|
||||
func newTestPicoChannel(t *testing.T) *PicoChannel {
|
||||
t.Helper()
|
||||
|
||||
|
|
@ -142,3 +150,380 @@ func (c *PicoChannel) addConnForTest(pc *picoConn) {
|
|||
}
|
||||
bySession[pc.id] = pc
|
||||
}
|
||||
|
||||
// --- HTTP message endpoint helpers ---
|
||||
|
||||
// newTestChannelWithConfig creates a PicoChannel and returns it alongside the bus.
|
||||
// The channel is NOT started — call Start to mark it running.
|
||||
func newTestChannelWithConfig(t *testing.T, token string, opts ...func(*config.PicoConfig)) (*PicoChannel, *bus.MessageBus) {
|
||||
t.Helper()
|
||||
b := bus.NewMessageBus()
|
||||
cfg := config.PicoConfig{Enabled: true}
|
||||
cfg.SetToken(token)
|
||||
for _, opt := range opts {
|
||||
opt(&cfg)
|
||||
}
|
||||
ch, err := NewPicoChannel(cfg, b)
|
||||
if err != nil {
|
||||
t.Fatalf("NewPicoChannel: %v", err)
|
||||
}
|
||||
return ch, b
|
||||
}
|
||||
|
||||
// newTestHTTPChannel creates a PicoChannel with default config for HTTP tests.
|
||||
func newTestHTTPChannel(t *testing.T) *PicoChannel {
|
||||
t.Helper()
|
||||
ch, _ := newTestChannelWithConfig(t, testToken)
|
||||
return ch
|
||||
}
|
||||
|
||||
// startTestChannel creates, starts, and registers cleanup for a PicoChannel.
|
||||
func startTestChannel(t *testing.T) *PicoChannel {
|
||||
t.Helper()
|
||||
ch := newTestHTTPChannel(t)
|
||||
if err := ch.Start(context.Background()); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { ch.Stop(context.Background()) })
|
||||
return ch
|
||||
}
|
||||
|
||||
// startTestChannelWithBus creates and starts a PicoChannel, returning the bus
|
||||
// for message verification.
|
||||
func startTestChannelWithBus(t *testing.T) (*PicoChannel, *bus.MessageBus) {
|
||||
t.Helper()
|
||||
ch, b := newTestChannelWithConfig(t, testToken)
|
||||
if err := ch.Start(context.Background()); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { ch.Stop(context.Background()) })
|
||||
return ch, b
|
||||
}
|
||||
|
||||
func authRequest(method, url, body, token string) *http.Request {
|
||||
req := httptest.NewRequest(method, url, strings.NewReader(body))
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
return req
|
||||
}
|
||||
|
||||
// consumeInbound reads one message from the bus with a short timeout.
|
||||
func consumeInbound(t *testing.T, b *bus.MessageBus) bus.InboundMessage {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
select {
|
||||
case msg := <-b.InboundChan():
|
||||
return msg
|
||||
case <-ctx.Done():
|
||||
t.Fatal("expected inbound message on bus, got none")
|
||||
return bus.InboundMessage{}
|
||||
}
|
||||
}
|
||||
|
||||
// --- HTTP message rejection tests ---
|
||||
|
||||
func TestHTTPMessage_MethodNotAllowed(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
for _, method := range []string{http.MethodGet, http.MethodPut, http.MethodDelete, http.MethodPatch} {
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(method, "/pico/message", `{"content":"hello"}`, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusMethodNotAllowed {
|
||||
t.Errorf("%s: expected %d, got %d", method, http.StatusMethodNotAllowed, rec.Code)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_NotRunning(t *testing.T) {
|
||||
ch := newTestHTTPChannel(t) // not started
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", `{"content":"hello"}`, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Errorf("expected %d, got %d", http.StatusServiceUnavailable, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_MissingAuth(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodPost, "/pico/message", strings.NewReader(`{"content":"hello"}`))
|
||||
// No Authorization header.
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Errorf("expected %d, got %d", http.StatusUnauthorized, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_InvalidAuth(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", `{"content":"hello"}`, "wrong-token")
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Errorf("expected %d, got %d", http.StatusUnauthorized, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_OversizedBody(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
oversized := bytes.Repeat([]byte("A"), maxHTTPBodySize+1)
|
||||
req := httptest.NewRequest(http.MethodPost, "/pico/message", bytes.NewReader(oversized))
|
||||
req.Header.Set("Authorization", "Bearer "+testToken)
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusRequestEntityTooLarge {
|
||||
t.Errorf("expected %d, got %d", http.StatusRequestEntityTooLarge, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_AcceptsMaxBodySize(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
// Exactly at the limit — should not trigger 413.
|
||||
// Body is not valid JSON, so expect 400 (not 413).
|
||||
body := bytes.Repeat([]byte("A"), maxHTTPBodySize)
|
||||
req := httptest.NewRequest(http.MethodPost, "/pico/message", bytes.NewReader(body))
|
||||
req.Header.Set("Authorization", "Bearer "+testToken)
|
||||
rec := httptest.NewRecorder()
|
||||
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code == http.StatusRequestEntityTooLarge {
|
||||
t.Error("body at exactly max size should not be rejected as too large")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_MalformedJSON(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", "not json at all", testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Errorf("expected %d, got %d", http.StatusBadRequest, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_EmptyContent(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", `{"content":""}`, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Errorf("expected %d, got %d", http.StatusBadRequest, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_WhitespaceContent(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", `{"content":" "}`, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Errorf("expected %d, got %d", http.StatusBadRequest, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// --- HTTP message acceptance tests ---
|
||||
|
||||
func TestHTTPMessage_Accepted(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
body := `{"content":"hello","session_id":"test-session"}`
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", body, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code)
|
||||
}
|
||||
|
||||
var resp httpMessageResponse
|
||||
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode response: %v", err)
|
||||
}
|
||||
if !resp.OK {
|
||||
t.Error("expected ok=true")
|
||||
}
|
||||
if resp.Status != "accepted" {
|
||||
t.Errorf("expected status=accepted, got %s", resp.Status)
|
||||
}
|
||||
if resp.SessionID != "test-session" {
|
||||
t.Errorf("expected session_id=test-session, got %s", resp.SessionID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_GeneratesSessionID(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", `{"content":"hello"}`, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code)
|
||||
}
|
||||
|
||||
var resp httpMessageResponse
|
||||
if err := json.NewDecoder(rec.Body).Decode(&resp); err != nil {
|
||||
t.Fatalf("decode response: %v", err)
|
||||
}
|
||||
if resp.SessionID == "" {
|
||||
t.Error("expected non-empty generated session_id")
|
||||
}
|
||||
}
|
||||
|
||||
// --- HTTP message bus semantics tests ---
|
||||
|
||||
func TestHTTPMessage_PublishesBusMessage(t *testing.T) {
|
||||
ch, b := startTestChannelWithBus(t)
|
||||
|
||||
body := `{"content":"hello world","session_id":"sess-42"}`
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", body, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code)
|
||||
}
|
||||
|
||||
msg := consumeInbound(t, b)
|
||||
|
||||
if msg.Channel != "pico" {
|
||||
t.Errorf("Channel: expected pico, got %s", msg.Channel)
|
||||
}
|
||||
if msg.Content != "hello world" {
|
||||
t.Errorf("Content: expected hello world, got %s", msg.Content)
|
||||
}
|
||||
if msg.ChatID != "pico:sess-42" {
|
||||
t.Errorf("ChatID: expected pico:sess-42, got %s", msg.ChatID)
|
||||
}
|
||||
if msg.Peer.Kind != "direct" {
|
||||
t.Errorf("Peer.Kind: expected direct, got %s", msg.Peer.Kind)
|
||||
}
|
||||
if msg.Peer.ID != "pico:sess-42" {
|
||||
t.Errorf("Peer.ID: expected pico:sess-42, got %s", msg.Peer.ID)
|
||||
}
|
||||
if msg.MessageID == "" {
|
||||
t.Error("MessageID: expected non-empty generated ID")
|
||||
}
|
||||
if msg.Metadata["platform"] != "pico" {
|
||||
t.Errorf("Metadata[platform]: expected pico, got %s", msg.Metadata["platform"])
|
||||
}
|
||||
if msg.Metadata["session_id"] != "sess-42" {
|
||||
t.Errorf("Metadata[session_id]: expected sess-42, got %s", msg.Metadata["session_id"])
|
||||
}
|
||||
if msg.Metadata["transport"] != "http" {
|
||||
t.Errorf("Metadata[transport]: expected http, got %s", msg.Metadata["transport"])
|
||||
}
|
||||
// SenderID is resolved to canonical form by BaseChannel.HandleMessage.
|
||||
if msg.SenderID != "pico:pico-user" {
|
||||
t.Errorf("SenderID: expected pico:pico-user, got %s", msg.SenderID)
|
||||
}
|
||||
if msg.Sender.Platform != "pico" {
|
||||
t.Errorf("Sender.Platform: expected pico, got %s", msg.Sender.Platform)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_PreservesContentWhitespace(t *testing.T) {
|
||||
ch, b := startTestChannelWithBus(t)
|
||||
|
||||
// Content has leading/trailing whitespace — should be preserved on the bus
|
||||
// (matching WebSocket message.send semantics).
|
||||
body := `{"content":" hello world ","session_id":"ws-compat"}`
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", body, testToken)
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Fatalf("expected %d, got %d", http.StatusAccepted, rec.Code)
|
||||
}
|
||||
|
||||
msg := consumeInbound(t, b)
|
||||
if msg.Content != " hello world " {
|
||||
t.Errorf("Content: expected ' hello world ', got %q", msg.Content)
|
||||
}
|
||||
}
|
||||
|
||||
// --- HTTP message auth variant tests ---
|
||||
|
||||
func TestHTTPMessage_QueryTokenAccepted(t *testing.T) {
|
||||
ch, _ := newTestChannelWithConfig(t, testToken, func(cfg *config.PicoConfig) {
|
||||
cfg.AllowTokenQuery = true
|
||||
})
|
||||
if err := ch.Start(context.Background()); err != nil {
|
||||
t.Fatalf("Start: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { ch.Stop(context.Background()) })
|
||||
|
||||
req := httptest.NewRequest(
|
||||
http.MethodPost,
|
||||
"/pico/message?token="+testToken,
|
||||
strings.NewReader(`{"content":"hello"}`),
|
||||
)
|
||||
rec := httptest.NewRecorder()
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Errorf("expected %d, got %d", http.StatusAccepted, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHTTPMessage_QueryTokenRejectedWhenDisabled(t *testing.T) {
|
||||
ch := startTestChannel(t) // AllowTokenQuery defaults to false
|
||||
|
||||
req := httptest.NewRequest(
|
||||
http.MethodPost,
|
||||
"/pico/message?token="+testToken,
|
||||
strings.NewReader(`{"content":"hello"}`),
|
||||
)
|
||||
rec := httptest.NewRecorder()
|
||||
ch.handleHTTPMessage(rec, req)
|
||||
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Errorf("expected %d, got %d", http.StatusUnauthorized, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// --- HTTP message routing tests ---
|
||||
|
||||
func TestServeHTTP_RoutesMessage(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/pico/message", `{"content":"routed"}`, testToken)
|
||||
ch.ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code != http.StatusAccepted {
|
||||
t.Errorf("expected %d, got %d", http.StatusAccepted, rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestServeHTTP_UnknownPathReturns404(t *testing.T) {
|
||||
ch := startTestChannel(t)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodGet, "/pico/unknown", nil)
|
||||
ch.ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code != http.StatusNotFound {
|
||||
t.Errorf("expected %d, got %d", http.StatusNotFound, rec.Code)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue