Pipeline request func

pipeline/ptinput/funcs/fn_http_request.go

@@ -1,17 +1,34 @@
 package funcs
 
 import (
+	"bytes"
+	"encoding/json"
 	"io"
+	"net"
 	"net/http"
+	"strconv"
+	"strings"
+	"time"
 
 	"github.com/GuanceCloud/platypus/pkg/ast"
 	"github.com/GuanceCloud/platypus/pkg/engine/runtime"
 	"github.com/GuanceCloud/platypus/pkg/errchain"
 )
 
+var defaultTransport http.RoundTripper = &http.Transport{
+	DialContext: ((&net.Dialer{
+		Timeout:   30 * time.Second,
+		KeepAlive: 30 * time.Second,
+	}).DialContext),
+	MaxIdleConns:          100,
+	IdleConnTimeout:       90 * time.Second,
+	TLSHandshakeTimeout:   10 * time.Second,
+	ExpectContinueTimeout: 1 * time.Second,
+}
+
 func HTTPRequestChecking(ctx *runtime.Context, funcExpr *ast.CallExpr) *errchain.PlError {
 	if err := reindexFuncArgs(funcExpr, []string{
-		"method", "url", "headers",
+		"method", "url", "headers", "body",
 	}, 2); err != nil {
 		return runtime.NewRunError(ctx, err.Error(), funcExpr.NamePos)
 	}
@@ -52,16 +69,31 @@ func HTTPRequest(ctx *runtime.Context, funcExpr *ast.CallExpr) *errchain.PlError
 		}
 	}
 
+	var reqBody io.Reader
+	if funcExpr.Param[3] != nil {
+		val, _, err := runtime.RunStmt(ctx, funcExpr.Param[3])
+		if err != nil {
+			return err
+		}
+		reqBody = buildBody(val)
+	}
+
 	// Send HTTP request
-	client := &http.Client{}
-	req, errR := http.NewRequest(method.(string), url.(string), nil)
+	client := &http.Client{
+		Transport: defaultTransport,
+		Timeout:   time.Duration(10) * time.Second,
+	}
+
+	req, errR := http.NewRequest(method.(string), url.(string), reqBody)
 	if errR != nil {
 		ctx.Regs.ReturnAppend(nil, ast.Nil)
 		return nil
 	}
 	if headers != nil {
 		for k, v := range headers.(map[string]any) {
-			req.Header.Set(k, v.(string))
+			if v, ok := v.(string); ok {
+				req.Header.Set(k, v)
+			}
 		}
 	}
 
@@ -87,3 +119,26 @@ func HTTPRequest(ctx *runtime.Context, funcExpr *ast.CallExpr) *errchain.PlError
 
 	return nil
 }
+
+func buildBody(val any) io.Reader {
+	switch val := val.(type) {
+	case string:
+		return strings.NewReader(val)
+	case []any:
+		if val, err := json.Marshal(val); err == nil {
+			return bytes.NewReader(val)
+		}
+	case map[string]any:
+		if val, err := json.Marshal(val); err == nil {
+			return bytes.NewReader(val)
+		}
+	case float64:
+		return strings.NewReader(strconv.FormatFloat(val, 'f', -1, 64))
+	case int64:
+		return strings.NewReader(strconv.FormatInt(val, 10))
+	case bool:
+		return strings.NewReader(strconv.FormatBool(val))
+	default:
+	}
+	return nil
+}

pipeline/ptinput/funcs/fn_http_request_test.go

@@ -2,17 +2,57 @@ package funcs
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 	"time"
 
 	"github.com/GuanceCloud/cliutils/pipeline/ptinput"
 	"github.com/GuanceCloud/cliutils/point"
-	tu "github.com/GuanceCloud/cliutils/testutil"
+	"github.com/stretchr/testify/assert"
 )
 
+func TestBuildBody(t *testing.T) {
+	cases := []struct {
+		val    any
+		result string
+	}{
+		{val: float64(123.1),
+			result: "123.1"},
+		{val: int64(123),
+			result: "123"},
+		{val: true,
+			result: "true"},
+		{val: false,
+			result: "false"},
+		{val: "abc",
+			result: "abc"},
+		{val: []any{1, 2, 3},
+			result: "[1,2,3]"},
+		{val: map[string]any{"a": 1, "b": 2},
+			result: `{"a":1,"b":2}`},
+		{val: nil,
+			result: ""},
+	}
+
+	for i, c := range cases {
+		t.Run(fmt.Sprintf("index_%d", i), func(t *testing.T) {
+			var buf []byte
+			if b := buildBody(c.val); b != nil {
+				var err error
+				buf, err = io.ReadAll(b)
+				if err != nil && !errors.Is(err, io.EOF) {
+					t.Error(err)
+				}
+			}
+			assert.Equal(t, c.result, string(buf))
+		})
+	}
+}
+
 func TestHTTPRequest(t *testing.T) {
 	server := HTTPServer()
 	defer server.Close()
@@ -27,30 +67,26 @@ func TestHTTPRequest(t *testing.T) {
 		outkey       string
 	}{
 		{
-			name: "acquire_code",
-			pl: `resp = http_request("GET", ` + url + `) 
-			add_key(abc, resp["status_code"])`,
+			name: "test_post",
+			pl: fmt.Sprintf(`
+			resp = http_request("POST", %s, {"extraHeader": "1", 
+			"extraHeader": "1"}, {"a": "1"})
+			add_key(abc, resp["body"])	
+			`, url),
 			in:       `[]`,
-			expected: int64(200),
 			outkey:   "abc",
+			expected: `{"a":"1"}`,
 		},
 		{
-			name: "acquire_body_without_headers",
-			pl: `resp = http_request("GET", ` + url + `)
-			resp_body = load_json(resp["body"])
-			add_key(abc, resp_body["a"])`,
+			name: "test_put",
+			pl: fmt.Sprintf(`
+			resp = http_request("put", %s, {"extraHeader": "1", 
+			"extraHeader": "1"}, {"a": "1"})
+			add_key(abc, resp["body"])	
+			`, url),
 			in:       `[]`,
-			expected: "hello",
-			outkey:   `abc`,
-		},
-		{
-			name: "acquire_body_with_headers",
-			pl: `resp = http_request("GET", ` + url + `, {"extraHeader1": "1", "extraHeader2": "1"})
-			resp_body = load_json(resp["body"])
-			add_key(abc, resp_body["a"])`,
-			in:       `[]`,
-			expected: "hello world",
-			outkey:   `abc`,
+			outkey:   "abc",
+			expected: `{"a":"1"}`,
 		},
 	}
 
@@ -75,7 +111,7 @@ func TestHTTPRequest(t *testing.T) {
 
 			v, _, _ := pt.Get(tc.outkey)
 			// tu.Equals(t, nil, err)
-			tu.Equals(t, tc.expected, v)
+			assert.Equal(t, tc.expected, v)
 
 			t.Logf("[%d] PASS", idx)
 		})
@@ -85,18 +121,31 @@ func TestHTTPRequest(t *testing.T) {
 func HTTPServer() *httptest.Server {
 	server := httptest.NewServer(http.HandlerFunc(
 		func(w http.ResponseWriter, r *http.Request) {
-			var responseData map[string]string
 			headers := r.Header
+
+			var respData []byte
+			var err error
 			if headers.Get("extraHeader1") != "" && headers.Get("extraHeader2") != "" {
-				responseData = map[string]string{"a": "hello world"}
+				responseData := map[string]string{"a": "hello world"}
+				respData, err = json.Marshal(responseData)
+				if err != nil {
+					http.Error(w, "Internal Server Error", http.StatusInternalServerError)
+				}
 			} else {
-				responseData = map[string]string{"a": "hello"}
-			}
-			responseJSON, err := json.Marshal(responseData)
-			if err != nil {
-				http.Error(w, "Internal Server Error", http.StatusInternalServerError)
+				switch r.Method {
+				case http.MethodGet:
+					responseData := map[string]string{"a": "hello"}
+					respData, err = json.Marshal(responseData)
+					if err != nil {
+						http.Error(w, "Internal Server Error", http.StatusInternalServerError)
+					}
+				default:
+					d, _ := io.ReadAll(r.Body)
+					respData = d
+				}
 			}
-			w.Write(responseJSON)
+
+			w.Write(respData)
 			w.WriteHeader(http.StatusOK)
 		},
 	))

pipeline/ptinput/funcs/md/http_request.en.md

@@ -1,6 +1,6 @@
 ### `http_request()` {#fn-http-request}
 
-Function prototype: `fn http_request(method: str, url: str, headers: map) map`
+Function prototype: `fn http_request(method: str, url: str, headers: map, body: any) map`
 
 Function description: Send an HTTP request, receive the response, and encapsulate it into a map
 
@@ -9,13 +9,14 @@ Function parameters:
 - `method`: GET|POST
 - `url`: Request path
 - `headers`: Additional header，the type is map[string]string
+- `body`: Request body
 
 Return type: map
 
 key contains status code (status_code) and result body (body)
 
-- `status_code`: status_code
-- `body`: response body
+- `status_code`: Status code
+- `body`: Response body
 
 Example:
 
@@ -25,4 +26,4 @@ resp_body = load_json(resp["body"])
 
 add_key(abc, resp["status_code"])
 add_key(abc, resp_body["a"])
-```
+```