85 lines
3.6 KiB
Go
85 lines
3.6 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"net/http"
|
||
"strings"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestIntegrationWorkflowQuerySDK(t *testing.T) {
|
||
api, queue, token := integrationWorkerClient(t)
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
spec := workflowSDKSpec(queue, "query."+workerTestID(t))
|
||
// SIGNAL保持稳定WAITING，不启动Worker或等待短时任务截止。
|
||
spec.Body = WorkflowBlock{Steps: []WorkflowStep{{Key: "receipt", Kind: "SIGNAL", Signal: &WorkflowSignal{Name: "done", PayloadVersion: "v1", Correlation: WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`"query"`)}, PayloadType: spec.InputType, TimeoutMS: 60000}}}, OutputType: spec.InputType, Output: WorkflowRef("receipt")}
|
||
in := publishWorkflowSDK(t, ctx, api, spec)
|
||
in.BusinessKey = "作业 /+%\n " + workerTestID(t) + " "
|
||
var started []WorkflowInstance
|
||
for _, key := range []string{in.BusinessKey, in.BusinessKey, strings.TrimSpace(in.BusinessKey)} {
|
||
in.BusinessKey = key
|
||
out, err := api.StartWorkflow(ctx, workerTestID(t), in)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
started = append(started, out.Instance)
|
||
}
|
||
paused, err := api.ControlWorkflow(ctx, started[0].ID, workerTestID(t), WorkflowControlRequest{Operation: "PAUSE", ExpectedControlRevision: started[0].ControlRevision, Reason: "verify paused query"})
|
||
if err != nil || !paused.Instance.Paused {
|
||
t.Fatal(paused, err)
|
||
}
|
||
from, before := started[0].CreatedAt, started[2].CreatedAt.Add(time.Microsecond)
|
||
query := ListWorkflowsRequest{DefinitionID: in.DefinitionID, Version: "v1", State: "WAITING", BusinessKey: started[0].BusinessKey, CreatedFrom: &from, CreatedBefore: &before, Limit: 1}
|
||
first, err := api.ListWorkflows(ctx, query)
|
||
if err != nil || len(first.Items) != 1 || first.Items[0].ID != started[1].ID || first.NextCursor == "" {
|
||
t.Fatal(first, err)
|
||
}
|
||
query.Cursor = first.NextCursor
|
||
last, err := api.ListWorkflows(ctx, query)
|
||
if err != nil || len(last.Items) != 1 || last.Items[0].ID != started[0].ID || !last.Items[0].Paused || last.NextCursor != "" {
|
||
t.Fatal(last, err)
|
||
}
|
||
query.BusinessKey = strings.TrimSpace(query.BusinessKey)
|
||
_, err = api.ListWorkflows(ctx, query)
|
||
workflowAPIError(t, err, "INVALID_ARGUMENT")
|
||
query.Cursor = ""
|
||
if page, err := api.ListWorkflows(ctx, query); err != nil || len(page.Items) != 1 || page.Items[0].ID != started[2].ID {
|
||
t.Fatal("exact business key", page, err)
|
||
}
|
||
b := false
|
||
query.BusinessKey = started[0].BusinessKey
|
||
query.Paused = &b
|
||
if page, err := api.ListWorkflows(ctx, query); err != nil || len(page.Items) != 1 || page.Items[0].ID != started[1].ID {
|
||
t.Fatal("false paused filter", page, err)
|
||
}
|
||
for _, suffix := range []string{"?state=WAITING&state=RUNNING", "?paused=1", "?business_key=", "?unknown=x", "?version=v1", "?limit=101"} {
|
||
req, err := http.NewRequestWithContext(ctx, "GET", api.baseURL+"/v1/workflows"+suffix, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
req.Header.Set("Authorization", "Bearer "+token)
|
||
res, err := http.DefaultClient.Do(req)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
res.Body.Close()
|
||
if res.StatusCode != 400 {
|
||
t.Fatal("invalid raw query", suffix, res.StatusCode)
|
||
}
|
||
}
|
||
for _, v := range started {
|
||
current, err := api.GetWorkflow(ctx, v.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err = api.ControlWorkflow(ctx, v.ID, workerTestID(t), WorkflowControlRequest{Operation: "CANCEL", ExpectedControlRevision: current.ControlRevision, Reason: "query regression cleanup"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
waitWorkflowSDK(t, ctx, api, v.ID, "CANCELLED")
|
||
}
|
||
t.Log("real HTTP: exact key, time bounds, fixed version, paused false, filtered cursor, malformed query rejection and cancellation cleanup")
|
||
}
|