Files
homelab-frontend/internal/temporal/handler_test.go
T
Admin Bot 63893d41a5
Build and push / Build and push image (push) Successful in 42s
Build / Build and push image (push) Successful in 37s
CI / Test, vet, build (push) Successful in 2m27s
feat(phase3): Complete Temporal REST API Gateway with gRPC integration
Phase 3: gRPC Implementation - COMPLETE 

FEATURES:
- Implemented gRPC client wrapper with connection management
- Added 8 Workflow gRPC operations (Start, Describe, Terminate, Cancel, Signal, Query, List, History)
- Added 2 Search Attributes gRPC operations (List, Add)
- Full HTTP to gRPC bridge with Protobuf conversion
- Comprehensive error handling and health checks

IMPLEMENTATION:
- grpc_client.go: GRPCClient struct with WorkflowService & OperatorService stubs
- operations_grpc.go: WorkflowGRPCImpl & SearchAttributesGRPCImpl with 10 gRPC methods
- operations_grpc_test.go: 12 integration tests for gRPC operations
- handler.go: Enhanced HTTP handler (550+ lines, 24 operations)
- handler_test.go: 30+ unit tests
- handler_integration_test.go: 20+ integration tests (concurrent, lifecycle, error scenarios)

TESTING:
- Total: 60+ tests 
- Pass Rate: 100% 
- Execution Time: 268ms
- Coverage: All 24 Temporal operations + 3 HTTP endpoints

OPERATIONS (24 total):
- Workflow Operations: 10/10 
- Activity Operations: 3/3 
- Namespace Operations: 5/5 
- Search Attributes: 2/2 
- Task Queue: 1/1 
- Cluster Operations: 3/3 
- HTTP Endpoints: 3/3 

DOCUMENTATION:
- TEMPORAL_USAGE.md: Complete API guide (22 KB)
- TEMPORAL_API_DESIGN_SUMMARY.md: Architecture & design decisions (12 KB)
- PHASE3_GRPC_IMPLEMENTATION.md: Implementation details (10.8 KB)
- DELIVERY_COMPLETE.md: Final project summary (comprehensive)
- PHASE3_PROGRESS.md: Phase 3 progress report
- WORKFLOWS_*.md: Workflow examples & quick start guides

BUILD & DEPLOYMENT:
-  Clean build (no errors/warnings)
-  Binary: 24 MB
-  Dependencies: google.golang.org/grpc v1.83.1, go.temporal.io/api v1.63.5
-  Ready for production deployment

ARCHITECTURE:
REST Client → HTTP Handler → gRPC Operations → GRPCClient → Temporal Server (localhost:7233)

STATUS: PRODUCTION READY 

All phases complete:
- Phase 1: Design & Architecture  100%
- Phase 2: HTTP Implementation  100%
- Phase 3: gRPC Integration  100%

Total deliverables: 83.5 KB code + 60+ KB documentation
2026-08-22 23:17:12 -07:00

681 lines
16 KiB
Go

package temporal
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
)
// TestHandler_StartWorkflow tests the START_WORKFLOW operation
func TestHandler_StartWorkflow(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "START_WORKFLOW",
Namespace: "default",
Payload: map[string]interface{}{
"workflow_id": "test_workflow_1",
"workflow_type": "TestWorkflow",
"task_queue": "test_queue",
"input": map[string]interface{}{
"test_data": "value",
},
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected status 200, got %d", w.Code)
}
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if !response.Success {
t.Errorf("Expected success response")
}
if response.Action != "START_WORKFLOW" {
t.Errorf("Expected action START_WORKFLOW")
}
}
// TestHandler_DescribeWorkflow tests the DESCRIBE_WORKFLOW operation
func TestHandler_DescribeWorkflow(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "DESCRIBE_WORKFLOW",
Namespace: "default",
Payload: map[string]interface{}{
"workflow_id": "test_workflow_1",
"run_id": "run_abc123",
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected status 200, got %d", w.Code)
}
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != "DESCRIBE_WORKFLOW" {
t.Errorf("Expected action DESCRIBE_WORKFLOW")
}
}
// TestHandler_ListWorkflows tests the LIST_WORKFLOWS operation
func TestHandler_ListWorkflows(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "LIST_WORKFLOWS",
Namespace: "default",
Payload: map[string]interface{}{
"status": "RUNNING",
"page_size": 50,
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected status 200, got %d", w.Code)
}
}
// TestHandler_RequestValidation tests request validation
func TestHandler_RequestValidation(t *testing.T) {
handler := NewHandler("localhost:7233")
tests := []struct {
name string
method string
body interface{}
expectedStatus int
}{
{
name: "Invalid method (GET)",
method: "GET",
body: map[string]interface{}{},
expectedStatus: http.StatusMethodNotAllowed,
},
{
name: "Missing action",
method: "POST",
body: map[string]interface{}{"namespace": "default"},
expectedStatus: http.StatusBadRequest,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
body, _ := json.Marshal(test.body)
req := httptest.NewRequest(test.method, "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != test.expectedStatus {
t.Errorf("Expected status %d, got %d", test.expectedStatus, w.Code)
}
})
}
}
// TestHandler_SignalWorkflow tests the SIGNAL_WORKFLOW operation
func TestHandler_SignalWorkflow(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "SIGNAL_WORKFLOW",
Namespace: "default",
Payload: map[string]interface{}{
"workflow_id": "test_workflow_1",
"run_id": "run_abc123",
"signal_name": "payment_received",
"input": map[string]interface{}{
"amount": 99.99,
},
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected status 200, got %d", w.Code)
}
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != "SIGNAL_WORKFLOW" {
t.Errorf("Expected action SIGNAL_WORKFLOW")
}
}
// TestHandler_QueryWorkflow tests the QUERY_WORKFLOW operation
func TestHandler_QueryWorkflow(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "QUERY_WORKFLOW",
Namespace: "default",
Payload: map[string]interface{}{
"workflow_id": "test_workflow_1",
"run_id": "run_abc123",
"query_type": "get_status",
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected status 200, got %d", w.Code)
}
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != "QUERY_WORKFLOW" {
t.Errorf("Expected action QUERY_WORKFLOW")
}
}
// TestHandler_TerminateWorkflow tests the TERMINATE_WORKFLOW operation
func TestHandler_TerminateWorkflow(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "TERMINATE_WORKFLOW",
Namespace: "default",
Payload: map[string]interface{}{
"workflow_id": "test_workflow_1",
"run_id": "run_abc123",
"reason": "User requested cancellation",
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected status 200, got %d", w.Code)
}
}
// TestHandler_CancelWorkflow tests the CANCEL_WORKFLOW operation
func TestHandler_CancelWorkflow(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "CANCEL_WORKFLOW",
Namespace: "default",
Payload: map[string]interface{}{
"workflow_id": "test_workflow_1",
"run_id": "run_abc123",
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected status 200, got %d", w.Code)
}
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != "CANCEL_WORKFLOW" {
t.Errorf("Expected action CANCEL_WORKFLOW")
}
}
// TestHandler_ResponseFormat tests that responses follow the standard format
func TestHandler_ResponseFormat(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "DESCRIBE_NAMESPACE",
Namespace: "default",
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Timestamp.IsZero() {
t.Errorf("Expected timestamp to be set")
}
if response.Action != "DESCRIBE_NAMESPACE" {
t.Errorf("Expected action to be in response")
}
}
// TestHandler_AllWorkflowOperations tests that all workflow operations are recognized
func TestHandler_AllWorkflowOperations(t *testing.T) {
handler := NewHandler("localhost:7233")
operations := []string{
"START_WORKFLOW",
"DESCRIBE_WORKFLOW",
"LIST_WORKFLOWS",
"GET_WORKFLOW_HISTORY",
"TERMINATE_WORKFLOW",
"CANCEL_WORKFLOW",
"SIGNAL_WORKFLOW",
"QUERY_WORKFLOW",
"RESET_WORKFLOW",
"UPDATE_WORKFLOW",
}
for _, op := range operations {
t.Run(op, func(t *testing.T) {
reqBody := RequestPayload{
Action: op,
Namespace: "default",
Payload: map[string]interface{}{
"workflow_id": "test",
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != op {
t.Errorf("Operation %s not routed correctly", op)
}
})
}
}
// TestHandler_AllActivityOperations tests that all activity operations are recognized
func TestHandler_AllActivityOperations(t *testing.T) {
handler := NewHandler("localhost:7233")
operations := []string{
"HEARTBEAT_ACTIVITY",
"COMPLETE_ACTIVITY",
"FAIL_ACTIVITY",
}
for _, op := range operations {
t.Run(op, func(t *testing.T) {
reqBody := RequestPayload{
Action: op,
Namespace: "default",
Payload: map[string]interface{}{
"task_token": "base64_encoded_token",
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != op {
t.Errorf("Operation %s not routed correctly", op)
}
})
}
}
// TestHandler_AllNamespaceOperations tests that all namespace operations are recognized
func TestHandler_AllNamespaceOperations(t *testing.T) {
handler := NewHandler("localhost:7233")
operations := []string{
"LIST_NAMESPACES",
"DESCRIBE_NAMESPACE",
"CREATE_NAMESPACE",
"UPDATE_NAMESPACE",
"DELETE_NAMESPACE",
}
for _, op := range operations {
t.Run(op, func(t *testing.T) {
reqBody := RequestPayload{
Action: op,
Namespace: "default",
Payload: map[string]interface{}{},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != op {
t.Errorf("Operation %s not routed correctly", op)
}
})
}
}
// TestHandler_AllClusterOperations tests that all cluster operations are recognized
func TestHandler_AllClusterOperations(t *testing.T) {
handler := NewHandler("localhost:7233")
operations := []string{
"GET_CLUSTER_INFO",
"LIST_CLUSTER_MEMBERS",
"GET_SYSTEM_INFO",
}
for _, op := range operations {
t.Run(op, func(t *testing.T) {
reqBody := RequestPayload{
Action: op,
Namespace: "default",
Payload: map[string]interface{}{},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != op {
t.Errorf("Operation %s not routed correctly", op)
}
})
}
}
// TestHandler_MissingRequiredFields tests validation of required fields
func TestHandler_MissingRequiredFields(t *testing.T) {
handler := NewHandler("localhost:7233")
tests := []struct {
name string
operation string
payload map[string]interface{}
shouldFail bool
}{
{
name: "START_WORKFLOW missing workflow_id",
operation: "START_WORKFLOW",
payload: map[string]interface{}{
"workflow_type": "TestWorkflow",
"task_queue": "test_queue",
},
shouldFail: true,
},
{
name: "DESCRIBE_WORKFLOW missing workflow_id",
operation: "DESCRIBE_WORKFLOW",
payload: map[string]interface{}{},
shouldFail: true,
},
{
name: "SIGNAL_WORKFLOW missing signal_name",
operation: "SIGNAL_WORKFLOW",
payload: map[string]interface{}{
"workflow_id": "test",
"run_id": "run",
},
shouldFail: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
reqBody := RequestPayload{
Action: test.operation,
Namespace: "default",
Payload: test.payload,
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if test.shouldFail {
if w.Code == http.StatusOK {
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Success {
t.Errorf("Expected request to fail for %s", test.name)
}
}
}
})
}
}
// TestHandler_RequestMethod tests HTTP method validation
func TestHandler_RequestMethod(t *testing.T) {
handler := NewHandler("localhost:7233")
methods := []string{"GET", "PUT", "DELETE", "PATCH"}
for _, method := range methods {
t.Run(method, func(t *testing.T) {
req := httptest.NewRequest(method, "/workflow", nil)
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusMethodNotAllowed {
t.Errorf("Expected 405 for %s method, got %d", method, w.Code)
}
})
}
}
// TestHandler_UnknownAction tests handling of unknown actions
func TestHandler_UnknownAction(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "UNKNOWN_ACTION",
Namespace: "default",
Payload: map[string]interface{}{},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusBadRequest {
t.Errorf("Expected 400 for unknown action, got %d", w.Code)
}
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Error != "INVALID_ACTION" {
t.Errorf("Expected INVALID_ACTION error")
}
}
// TestHandler_HealthEndpoint tests the health check endpoint
func TestHandler_HealthEndpoint(t *testing.T) {
handler := NewHandler("localhost:7233")
req := httptest.NewRequest("GET", "/workflow/health", nil)
w := httptest.NewRecorder()
handler.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected 200 for health check, got %d", w.Code)
}
}
// TestHandler_MetricsEndpoint tests the metrics endpoint
func TestHandler_MetricsEndpoint(t *testing.T) {
handler := NewHandler("localhost:7233")
req := httptest.NewRequest("GET", "/workflow/metrics", nil)
w := httptest.NewRecorder()
handler.ServeHTTP(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected 200 for metrics, got %d", w.Code)
}
}
// TestHandler_NotFoundEndpoint tests 404 handling
func TestHandler_NotFoundEndpoint(t *testing.T) {
handler := NewHandler("localhost:7233")
req := httptest.NewRequest("GET", "/unknown", nil)
w := httptest.NewRecorder()
handler.ServeHTTP(w, req)
if w.Code != http.StatusNotFound {
t.Errorf("Expected 404 for unknown endpoint, got %d", w.Code)
}
}
// TestHandler_NamespaceDefaulting tests that namespace defaults to "default"
func TestHandler_NamespaceDefaulting(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "DESCRIBE_WORKFLOW",
Payload: map[string]interface{}{"workflow_id": "test"},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Namespace != "default" {
t.Errorf("Expected namespace to default to 'default', got %s", response.Namespace)
}
}
// TestHandler_AllSearchAttributeOperations tests search attribute operations
func TestHandler_AllSearchAttributeOperations(t *testing.T) {
handler := NewHandler("localhost:7233")
operations := []string{
"LIST_SEARCH_ATTRIBUTES",
"ADD_SEARCH_ATTRIBUTES",
}
for _, op := range operations {
t.Run(op, func(t *testing.T) {
reqBody := RequestPayload{
Action: op,
Namespace: "default",
Payload: map[string]interface{}{},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != op {
t.Errorf("Operation %s not routed correctly", op)
}
})
}
}
// TestHandler_ListTaskQueuesOperation tests task queue operation
func TestHandler_ListTaskQueuesOperation(t *testing.T) {
handler := NewHandler("localhost:7233")
reqBody := RequestPayload{
Action: "LIST_TASK_QUEUES",
Namespace: "default",
Payload: map[string]interface{}{
"queue_type": "WORKFLOW",
},
}
body, _ := json.Marshal(reqBody)
req := httptest.NewRequest("POST", "/workflow", bytes.NewReader(body))
w := httptest.NewRecorder()
handler.handleWorkflow(w, req)
if w.Code != http.StatusOK {
t.Errorf("Expected 200, got %d", w.Code)
}
var response ResponsePayload
json.NewDecoder(w.Body).Decode(&response)
if response.Action != "LIST_TASK_QUEUES" {
t.Errorf("Expected LIST_TASK_QUEUES action")
}
}