From b0f608f145bc53f5aec0b926e1a93487b0173de4 Mon Sep 17 00:00:00 2001 From: Admin Bot Date: Mon, 14 Sep 2026 08:18:30 +0900 Subject: [PATCH] refactor: retire /workflows endpoint, use X-Service: workflow instead Remove deprecated /workflows HTTP endpoint in favor of unified X-Service header routing. All workflow operations now route through: X-Service: workflow X-Resource: {action} (start, describe, signal, query, etc) This consolidates routing patterns and allows users to specify domain/ namespace via request payload instead of path prefixes. Changes: - Remove internal/proxy/workflows.go (484 lines of predefined workflows) - Remove internal/proxy/workflows_test.go - Remove /workflows handler from proxy.ServeHTTP() - Update README.md to document X-Service routing pattern - Add migration note: use X-Service: workflow instead of /workflows WorkflowAdapter in serviceadapter/ handles X-Service: workflow requests and forwards to Temporal gRPC API. Users can now specify domain/namespace in request payload for multi-tenant workflow access. Related: https://forgejo.riotpiao.com/riotpiao-poimen/homelab-frontend/pulls/26 --- README.md | 57 ++-- internal/proxy/proxy.go | 9 +- internal/proxy/workflows.go | 484 ------------------------------- internal/proxy/workflows_test.go | 258 ---------------- 4 files changed, 30 insertions(+), 778 deletions(-) delete mode 100644 internal/proxy/workflows.go delete mode 100644 internal/proxy/workflows_test.go diff --git a/README.md b/README.md index 8ff71c8..c47925f 100644 --- a/README.md +++ b/README.md @@ -38,22 +38,22 @@ Production API gateway for the homelab cluster. Single entry point (`api.riotpia │ (routing, auth, limits) │ └──────┬───────────────────────┘ │ - ┌──────┴──────────────────────────────────┐ - │ │ - /v1/* /workflow /sqs / -(LLM) (Temporal gRPC) (Queues) (X-Service) - │ │ │ │ - ▼ ▼ ▼ ▼ -llm-serving temporal:7233 kmsvc/Kafka IAM, S3 -(vLLM, Ollama) (WorkflowService) Memory -(TEI) (gRPC bridge) (poimen) + ┌──────┴───────────────────────────────────┐ + │ │ + /v1/* X-Service header routing / +(LLM) (workflow, sqs, s3, iam, memory) / + │ │ │ + ▼ ▼ ▼ +llm-serving temporal:7233 kmsvc/Kafka, MinIO, +(vLLM, Ollama) (gRPC) Authentik, poimen-memory +(TEI) ``` **Design principles:** -- ✅ Single hostname, multiple path prefixes -- ✅ HTTP REST gateway → gRPC Temporal bridge +- ✅ Single hostname, unified X-Service + X-Resource header routing +- ✅ HTTP REST gateway → gRPC Temporal bridge (via X-Service: workflow) - ✅ Bearer token auth via Authentik (JWT + RBAC) -- ✅ Streaming unbuffered (SSE, WebSocket) +- ✅ Streaming unbuffered (SSE, WebSocket, HTTP/2 multiplexing) - ✅ Per-route timeouts & rate limits - ✅ No cluster credentials held by gateway @@ -61,16 +61,16 @@ llm-serving temporal:7233 kmsvc/Kafka IAM, S3 ## Services & Capabilities -| Service | Prefix | Upstream | Status | +| Service | Method | Upstream | Status | |---------|--------|----------|--------| -| **LLM Chat** | `/v1/chat/completions` | llm-serving (vLLM) | ✅ Live | -| **Embeddings** | `/v1/embeddings` | llm-serving (TEI) | ✅ Live | -| **Reranking** | `/v1/rerank` | llm-serving (TEI) | ✅ Live | -| **Workflows** | `/workflow` | Temporal gRPC (7233) | ✅ Live (START, DESCRIBE, SIGNAL, QUERY, etc) | -| **Queues** | `/` + `X-Service: sqs` | kmsvc/Kafka | ⏳ Ready (ServiceAdapter) | -| **Memory** | `/` + `X-Service: memory` | poimen-memory | ✅ Live | -| **IAM** | `/` + `X-Service: iam` | Authentik API | ✅ Live | -| **S3** | `/` + `X-Service: s3` | MinIO | ✅ Live | +| **LLM Chat** | `POST /v1/chat/completions` | llm-serving (vLLM) | ✅ Live | +| **Embeddings** | `POST /v1/embeddings` | llm-serving (TEI) | ✅ Live | +| **Reranking** | `POST /v1/rerank` | llm-serving (TEI) | ✅ Live | +| **Workflows** | `X-Service: workflow` + `X-Resource: {action}` | Temporal gRPC (7233) | ✅ Live (START, DESCRIBE, SIGNAL, QUERY, etc) | +| **Queues** | `X-Service: sqs` + `X-Resource: {action}` | kmsvc/Kafka | ✅ Live | +| **Memory** | `X-Service: memory` + `X-Resource: {action}` | poimen-memory | ✅ Live | +| **IAM** | `X-Service: iam` + `X-Resource: {action}` | Authentik API | ✅ Live | +| **S3** | `X-Service: s3` + `X-Resource: {action}` | MinIO | ✅ Live | --- @@ -102,18 +102,17 @@ curl -X POST https://api.riotpiao.com/v1/chat/completions \ }' ``` -**Workflow:** +**Workflow (via X-Service header):** ```bash -curl -X POST https://api.riotpiao.com/workflow \ +curl -X POST https://api.riotpiao.com/ \ -H "Authorization: Bearer $TOKEN" \ + -H "X-Service: workflow" \ + -H "X-Resource: start" \ -d '{ - "action": "START_WORKFLOW", "namespace": "default", - "payload": { - "workflow_id": "my-workflow", - "workflow_type": "MyWorkflow", - "task_queue": "default" - } + "workflow_id": "my-workflow", + "workflow_type": "MyWorkflow", + "task_queue": "default" }' ``` diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 9373558..66fddc5 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -271,13 +271,8 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - // Handle /workflows endpoint (workflow orchestration) - if r.URL.Path == "/workflows" { - h.handleWorkflow(w, r) - return - } - - // Try to find a matching route (including body-based dispatch for /v1/chat/completions) + // Try to find a matching route (including body-based dispatch for /v1/chat/completions). + // Note: /workflows endpoint is deprecated. Use X-Service: workflow + X-Resource headers instead. route, err := h.RouteRequest(r) // Check if this is a model validation error (from body-based dispatch) diff --git a/internal/proxy/workflows.go b/internal/proxy/workflows.go deleted file mode 100644 index 88c2bfb..0000000 --- a/internal/proxy/workflows.go +++ /dev/null @@ -1,484 +0,0 @@ -// Package proxy provides request routing and forwarding. -package proxy - -import ( - "bytes" - "context" - "encoding/json" - "fmt" - "io" - "net/http" - "time" -) - -// WorkflowRequest represents a workflow execution request -type WorkflowRequest struct { - // Workflow ID or name - Workflow string `json:"workflow"` - - // Input parameters for the workflow - Input map[string]interface{} `json:"input"` - - // Optional: timeout in seconds - Timeout int `json:"timeout,omitempty"` - - // Optional: wait for result (default: true) - Wait *bool `json:"wait,omitempty"` -} - -// WorkflowResponse represents the response from workflow execution -type WorkflowResponse struct { - // Workflow execution ID - ID string `json:"id"` - - // Workflow name - Workflow string `json:"workflow"` - - // Execution status: pending, running, completed, failed - Status string `json:"status"` - - // Output of the workflow - Output interface{} `json:"output,omitempty"` - - // Error message if workflow failed - Error string `json:"error,omitempty"` - - // Timestamp when workflow was created - CreatedAt time.Time `json:"created_at"` - - // Timestamp when workflow completed - CompletedAt *time.Time `json:"completed_at,omitempty"` -} - -// PredefinedWorkflow defines a workflow template that combines multiple API calls -type PredefinedWorkflow struct { - Name string - Description string - Handler func(*http.Request, *Handler, map[string]interface{}) (interface{}, error) -} - -// handleWorkflow handles the /workflows endpoint -// It accepts workflow definitions and orchestrates API calls -func (h *Handler) handleWorkflow(w http.ResponseWriter, r *http.Request) { - // Only POST is supported - if r.Method != "POST" { - w.Header().Set("Content-Type", "application/problem+json") - w.WriteHeader(http.StatusMethodNotAllowed) - fmt.Fprintf(w, `{"type":"https://api.example.com/problems/method-not-allowed","title":"Method Not Allowed","status":405,"detail":"Only POST is supported for /workflows"}`) - return - } - - // Parse request body - var workflowReq WorkflowRequest - if err := json.NewDecoder(r.Body).Decode(&workflowReq); err != nil { - writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/invalid-workflow-request", "Invalid Workflow Request", "Failed to parse workflow request: "+err.Error(), nil) - return - } - - // Validate workflow name - if workflowReq.Workflow == "" { - writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/missing-workflow", "Missing Workflow", "The 'workflow' field is required", nil) - return - } - - // Get predefined workflow - workflow, ok := h.getWorkflow(workflowReq.Workflow) - if !ok { - availableWorkflows := h.getAvailableWorkflows() - writeProblemDetail(w, http.StatusBadRequest, "https://api.example.com/problems/unknown-workflow", "Unknown Workflow", fmt.Sprintf("Workflow %q is not available", workflowReq.Workflow), availableWorkflows) - return - } - - // Default wait to true - wait := true - if workflowReq.Wait != nil { - wait = *workflowReq.Wait - } - - // Set default timeout if not provided - timeout := time.Duration(30) * time.Second - if workflowReq.Timeout > 0 { - timeout = time.Duration(workflowReq.Timeout) * time.Second - } - - // Create a context with timeout for workflow execution - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - - // Execute workflow - output, err := workflow.Handler(r.WithContext(ctx), h, workflowReq.Input) - - // Build response - workflowResp := WorkflowResponse{ - ID: generateWorkflowID(), - Workflow: workflowReq.Workflow, - CreatedAt: time.Now(), - } - - if err != nil { - workflowResp.Status = "failed" - workflowResp.Error = err.Error() - } else { - if wait { - workflowResp.Status = "completed" - workflowResp.Output = output - now := time.Now() - workflowResp.CompletedAt = &now - } else { - workflowResp.Status = "pending" - } - } - - // Write response - w.Header().Set("Content-Type", "application/json") - if err != nil { - w.WriteHeader(http.StatusInternalServerError) - } else { - w.WriteHeader(http.StatusOK) - } - json.NewEncoder(w).Encode(workflowResp) -} - -// getWorkflow returns a predefined workflow by name -func (h *Handler) getWorkflow(name string) (*PredefinedWorkflow, bool) { - workflows := h.getPredefinedWorkflows() - for _, wf := range workflows { - if wf.Name == name { - return &wf, true - } - } - return nil, false -} - -// getPredefinedWorkflows returns all available workflows -func (h *Handler) getPredefinedWorkflows() []PredefinedWorkflow { - return []PredefinedWorkflow{ - { - Name: "chat-and-embed", - Description: "Chat with a model and then embed the response", - Handler: h.chatAndEmbedWorkflow, - }, - { - Name: "multi-model-chat", - Description: "Chat with multiple models sequentially", - Handler: h.multiModelChatWorkflow, - }, - { - Name: "rag-pipeline", - Description: "RAG pipeline: embed query, rerank, then chat with context", - Handler: h.ragPipelineWorkflow, - }, - { - Name: "batch-embeddings", - Description: "Generate embeddings for multiple texts", - Handler: h.batchEmbeddingsWorkflow, - }, - } -} - -// getAvailableWorkflows returns a list of available workflow names -func (h *Handler) getAvailableWorkflows() []string { - workflows := h.getPredefinedWorkflows() - names := make([]string, len(workflows)) - for i, wf := range workflows { - names[i] = wf.Name - } - return names -} - -// Workflow implementations - -// chatAndEmbedWorkflow: Chat with a model, then embed the response -func (h *Handler) chatAndEmbedWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) { - model, ok := input["model"].(string) - if !ok || model == "" { - return nil, fmt.Errorf("missing required parameter: model") - } - - embedModel, ok := input["embed_model"].(string) - if !ok { - embedModel = "nomic-ai/nomic-embed-text-v2-moe" - } - - messages, ok := input["messages"].([]interface{}) - if !ok { - return nil, fmt.Errorf("missing required parameter: messages") - } - - // Step 1: Chat - chatReq := map[string]interface{}{ - "model": model, - "messages": messages, - } - - chatBody, _ := json.Marshal(chatReq) - chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody))) - chatHTTPReq.Header.Set("Content-Type", "application/json") - - // Create a response writer to capture the chat response - chatResp := &responseCapture{} - handler.ServeHTTP(chatResp, chatHTTPReq) - - var chatResult map[string]interface{} - if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil { - return nil, fmt.Errorf("failed to parse chat response: %v", err) - } - - // Extract message content - var messageContent string - if choices, ok := chatResult["choices"].([]interface{}); ok && len(choices) > 0 { - if choice, ok := choices[0].(map[string]interface{}); ok { - if message, ok := choice["message"].(map[string]interface{}); ok { - if content, ok := message["content"].(string); ok { - messageContent = content - } - } - } - } - - // Step 2: Embed the response - embedReq := map[string]interface{}{ - "model": embedModel, - "input": messageContent, - } - - embedBody, _ := json.Marshal(embedReq) - embedHTTPReq, _ := http.NewRequest("POST", "/v1/embeddings", io.NopCloser(bytes.NewReader(embedBody))) - embedHTTPReq.Header.Set("Content-Type", "application/json") - - embedResp := &responseCapture{} - handler.ServeHTTP(embedResp, embedHTTPReq) - - var embedResult map[string]interface{} - if err := json.Unmarshal(embedResp.body.Bytes(), &embedResult); err != nil { - return nil, fmt.Errorf("failed to parse embedding response: %v", err) - } - - return map[string]interface{}{ - "chat_response": chatResult, - "embedding_response": embedResult, - }, nil -} - -// multiModelChatWorkflow: Chat with multiple models sequentially -func (h *Handler) multiModelChatWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) { - models, ok := input["models"].([]interface{}) - if !ok || len(models) == 0 { - return nil, fmt.Errorf("missing required parameter: models (array)") - } - - messages, ok := input["messages"].([]interface{}) - if !ok { - return nil, fmt.Errorf("missing required parameter: messages") - } - - results := make([]map[string]interface{}, 0) - - for _, modelInterface := range models { - model, ok := modelInterface.(string) - if !ok { - continue - } - - chatReq := map[string]interface{}{ - "model": model, - "messages": messages, - } - - chatBody, _ := json.Marshal(chatReq) - chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody))) - chatHTTPReq.Header.Set("Content-Type", "application/json") - - chatResp := &responseCapture{} - handler.ServeHTTP(chatResp, chatHTTPReq) - - var chatResult map[string]interface{} - if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil { - results = append(results, map[string]interface{}{ - "model": model, - "error": err.Error(), - }) - continue - } - - results = append(results, map[string]interface{}{ - "model": model, - "result": chatResult, - }) - } - - return results, nil -} - -// ragPipelineWorkflow: RAG pipeline - embed query, rerank, chat with context -func (h *Handler) ragPipelineWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) { - query, ok := input["query"].(string) - if !ok || query == "" { - return nil, fmt.Errorf("missing required parameter: query") - } - - documents, ok := input["documents"].([]interface{}) - if !ok { - return nil, fmt.Errorf("missing required parameter: documents") - } - - model, ok := input["model"].(string) - if !ok { - model = "reasoning" - } - - rerankModel, ok := input["rerank_model"].(string) - if !ok { - rerankModel = "BAAI/bge-reranker-base" - } - - topK := 3 - if tk, ok := input["top_k"].(float64); ok { - topK = int(tk) - } - - // Step 1: Rerank documents based on query - rerankReq := map[string]interface{}{ - "model": rerankModel, - "query": query, - "texts": documents, - "top_k": topK, - } - - rerankBody, _ := json.Marshal(rerankReq) - rerankHTTPReq, _ := http.NewRequest("POST", "/v1/rerank", io.NopCloser(bytes.NewReader(rerankBody))) - rerankHTTPReq.Header.Set("Content-Type", "application/json") - - rerankResp := &responseCapture{} - handler.ServeHTTP(rerankResp, rerankHTTPReq) - - var rerankResult map[string]interface{} - if err := json.Unmarshal(rerankResp.body.Bytes(), &rerankResult); err != nil { - return nil, fmt.Errorf("failed to parse rerank response: %v", err) - } - - // Extract top documents - var topDocs []string - if results, ok := rerankResult["results"].([]interface{}); ok { - for i, resultInterface := range results { - if i >= topK { - break - } - if result, ok := resultInterface.(map[string]interface{}); ok { - if text, ok := result["text"].(string); ok { - topDocs = append(topDocs, text) - } - } - } - } - - // Step 2: Chat with context - context := fmt.Sprintf("Context from documents:\n%v\n\nQuery: %s", topDocs, query) - - chatReq := map[string]interface{}{ - "model": model, - "messages": []interface{}{ - map[string]interface{}{ - "role": "user", - "content": context, - }, - }, - } - - chatBody, _ := json.Marshal(chatReq) - chatHTTPReq, _ := http.NewRequest("POST", "/v1/chat/completions", io.NopCloser(bytes.NewReader(chatBody))) - chatHTTPReq.Header.Set("Content-Type", "application/json") - - chatResp := &responseCapture{} - handler.ServeHTTP(chatResp, chatHTTPReq) - - var chatResult map[string]interface{} - if err := json.Unmarshal(chatResp.body.Bytes(), &chatResult); err != nil { - return nil, fmt.Errorf("failed to parse chat response: %v", err) - } - - return map[string]interface{}{ - "reranked_documents": topDocs, - "chat_response": chatResult, - }, nil -} - -// batchEmbeddingsWorkflow: Generate embeddings for multiple texts -func (h *Handler) batchEmbeddingsWorkflow(r *http.Request, handler *Handler, input map[string]interface{}) (interface{}, error) { - texts, ok := input["texts"].([]interface{}) - if !ok || len(texts) == 0 { - return nil, fmt.Errorf("missing required parameter: texts (array)") - } - - model, ok := input["model"].(string) - if !ok { - model = "nomic-ai/nomic-embed-text-v2-moe" - } - - // Convert interface{} to []string - textStrings := make([]string, 0) - for _, t := range texts { - if str, ok := t.(string); ok { - textStrings = append(textStrings, str) - } - } - - if len(textStrings) == 0 { - return nil, fmt.Errorf("no valid text strings in texts array") - } - - embedReq := map[string]interface{}{ - "model": model, - "input": textStrings, - } - - embedBody, _ := json.Marshal(embedReq) - embedHTTPReq, _ := http.NewRequest("POST", "/v1/embeddings", io.NopCloser(bytes.NewReader(embedBody))) - embedHTTPReq.Header.Set("Content-Type", "application/json") - - embedResp := &responseCapture{} - handler.ServeHTTP(embedResp, embedHTTPReq) - - var embedResult map[string]interface{} - if err := json.Unmarshal(embedResp.body.Bytes(), &embedResult); err != nil { - return nil, fmt.Errorf("failed to parse embedding response: %v", err) - } - - return embedResult, nil -} - -// Utility functions - -// responseCapture captures HTTP response for reuse within workflows -type responseCapture struct { - status int - header http.Header - body bytes.Buffer -} - -func (w *responseCapture) Header() http.Header { - if w.header == nil { - w.header = make(http.Header) - } - return w.header -} - -func (w *responseCapture) Write(b []byte) (int, error) { - if w.status == 0 { - w.status = http.StatusOK - } - return w.body.Write(b) -} - -func (w *responseCapture) WriteHeader(statusCode int) { - if w.status == 0 { - w.status = statusCode - } -} - -// generateWorkflowID generates a unique workflow execution ID -func generateWorkflowID() string { - return fmt.Sprintf("wf_%d", time.Now().UnixNano()) -} - - diff --git a/internal/proxy/workflows_test.go b/internal/proxy/workflows_test.go deleted file mode 100644 index 8135f25..0000000 --- a/internal/proxy/workflows_test.go +++ /dev/null @@ -1,258 +0,0 @@ -package proxy - -import ( - "bytes" - "encoding/json" - "net/http" - "net/http/httptest" - "testing" - - "forgejo.riotpiao.com/rock/homelab-frontend/internal/config" -) - -func TestWorkflowEndpointNotFound(t *testing.T) { - // Create a minimal config - cfg := &config.Config{ - Routes: make(map[string]*config.Route), - Models: map[string]*config.ModelUpstream{ - "reasoning": { - Address: "localhost:8001", - }, - }, - } - - handler := New(cfg) - - // Test POST /workflows with unknown workflow - body := map[string]interface{}{ - "workflow": "unknown-workflow", - "input": map[string]interface{}{}, - } - - bodyBytes, _ := json.Marshal(body) - req := httptest.NewRequest("POST", "/workflows", bytes.NewReader(bodyBytes)) - req.Header.Set("Content-Type", "application/json") - - w := httptest.NewRecorder() - handler.ServeHTTP(w, req) - - if w.Code != http.StatusBadRequest { - t.Errorf("Expected 400, got %d", w.Code) - } - - var response map[string]interface{} - json.Unmarshal(w.Body.Bytes(), &response) - - if response["type"] != "https://api.example.com/problems/unknown-workflow" { - t.Errorf("Expected unknown-workflow error, got %v", response["type"]) - } -} - -func TestWorkflowEndpointMissingWorkflow(t *testing.T) { - cfg := &config.Config{ - Routes: make(map[string]*config.Route), - Models: make(map[string]*config.ModelUpstream), - } - - handler := New(cfg) - - // Test POST /workflows with missing workflow field - body := map[string]interface{}{ - "input": map[string]interface{}{}, - } - - bodyBytes, _ := json.Marshal(body) - req := httptest.NewRequest("POST", "/workflows", bytes.NewReader(bodyBytes)) - req.Header.Set("Content-Type", "application/json") - - w := httptest.NewRecorder() - handler.ServeHTTP(w, req) - - if w.Code != http.StatusBadRequest { - t.Errorf("Expected 400, got %d", w.Code) - } - - var response map[string]interface{} - json.Unmarshal(w.Body.Bytes(), &response) - - if response["type"] != "https://api.example.com/problems/missing-workflow" { - t.Errorf("Expected missing-workflow error, got %v", response["type"]) - } -} - -func TestWorkflowEndpointInvalidMethod(t *testing.T) { - cfg := &config.Config{ - Routes: make(map[string]*config.Route), - Models: make(map[string]*config.ModelUpstream), - } - - handler := New(cfg) - - // Test GET /workflows (should be 405) - req := httptest.NewRequest("GET", "/workflows", nil) - w := httptest.NewRecorder() - handler.ServeHTTP(w, req) - - if w.Code != http.StatusMethodNotAllowed { - t.Errorf("Expected 405, got %d", w.Code) - } -} - -func TestWorkflowEndpointInvalidJSON(t *testing.T) { - cfg := &config.Config{ - Routes: make(map[string]*config.Route), - Models: make(map[string]*config.ModelUpstream), - } - - handler := New(cfg) - - // Test POST /workflows with invalid JSON - req := httptest.NewRequest("POST", "/workflows", bytes.NewReader([]byte("not json"))) - req.Header.Set("Content-Type", "application/json") - - w := httptest.NewRecorder() - handler.ServeHTTP(w, req) - - if w.Code != http.StatusBadRequest { - t.Errorf("Expected 400, got %d", w.Code) - } -} - -func TestGetAvailableWorkflows(t *testing.T) { - cfg := &config.Config{ - Routes: make(map[string]*config.Route), - Models: make(map[string]*config.ModelUpstream), - } - - handler := New(cfg) - - workflows := handler.getAvailableWorkflows() - - expectedWorkflows := []string{ - "chat-and-embed", - "multi-model-chat", - "rag-pipeline", - "batch-embeddings", - } - - if len(workflows) != len(expectedWorkflows) { - t.Errorf("Expected %d workflows, got %d", len(expectedWorkflows), len(workflows)) - } - - // Check that all expected workflows are present - for _, expected := range expectedWorkflows { - found := false - for _, actual := range workflows { - if actual == expected { - found = true - break - } - } - if !found { - t.Errorf("Expected workflow %q not found", expected) - } - } -} - -func TestGetWorkflow(t *testing.T) { - cfg := &config.Config{ - Routes: make(map[string]*config.Route), - Models: make(map[string]*config.ModelUpstream), - } - - handler := New(cfg) - - // Test getting a valid workflow - workflow, ok := handler.getWorkflow("chat-and-embed") - if !ok { - t.Error("Expected to find chat-and-embed workflow") - } - if workflow.Name != "chat-and-embed" { - t.Errorf("Expected workflow name chat-and-embed, got %s", workflow.Name) - } - - // Test getting an invalid workflow - workflow, ok = handler.getWorkflow("invalid-workflow") - if ok { - t.Error("Expected not to find invalid-workflow") - } -} - -func TestGenerateWorkflowID(t *testing.T) { - id1 := generateWorkflowID() - id2 := generateWorkflowID() - - if id1 == id2 { - t.Error("Generated workflow IDs should be unique") - } - - if !bytes.HasPrefix([]byte(id1), []byte("wf_")) { - t.Errorf("Workflow ID should start with 'wf_', got %s", id1) - } -} - -func TestResponseCapture(t *testing.T) { - rc := &responseCapture{} - - // Test Header - rc.Header().Set("X-Test", "value") - if rc.Header().Get("X-Test") != "value" { - t.Error("Header not set correctly") - } - - // Test Write - n, err := rc.Write([]byte("test content")) - if err != nil { - t.Errorf("Unexpected error: %v", err) - } - if n != 12 { - t.Errorf("Expected 12 bytes written, got %d", n) - } - if rc.body.String() != "test content" { - t.Errorf("Expected 'test content', got %s", rc.body.String()) - } - - // Test WriteHeader - rc.WriteHeader(http.StatusOK) - if rc.status != http.StatusOK { - t.Errorf("Expected status 200, got %d", rc.status) - } - - // Test WriteHeader doesn't override - rc.WriteHeader(http.StatusInternalServerError) - if rc.status != http.StatusOK { - t.Error("WriteHeader should not override existing status") - } -} - -func TestWorkflowResponseSerialization(t *testing.T) { - resp := WorkflowResponse{ - ID: "wf_123", - Workflow: "test-workflow", - Status: "completed", - Output: map[string]interface{}{ - "key": "value", - }, - Error: "", - } - - data, err := json.Marshal(resp) - if err != nil { - t.Errorf("Failed to marshal response: %v", err) - } - - var unmarshaled WorkflowResponse - if err := json.Unmarshal(data, &unmarshaled); err != nil { - t.Errorf("Failed to unmarshal response: %v", err) - } - - if unmarshaled.ID != resp.ID { - t.Errorf("Expected ID %s, got %s", resp.ID, unmarshaled.ID) - } - if unmarshaled.Workflow != resp.Workflow { - t.Errorf("Expected Workflow %s, got %s", resp.Workflow, unmarshaled.Workflow) - } - if unmarshaled.Status != resp.Status { - t.Errorf("Expected Status %s, got %s", resp.Status, unmarshaled.Status) - } -}