110 lines
4.7 KiB
Go
110 lines
4.7 KiB
Go
package orbit
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"io"
|
||
"net/http"
|
||
"net/http/httptest"
|
||
"net/http/httputil"
|
||
"net/url"
|
||
"reflect"
|
||
"strings"
|
||
"sync/atomic"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
func TestIntegrationWorkflowDefinitionSDK(t *testing.T) {
|
||
api, queue, _ := integrationWorkerClient(t)
|
||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||
defer cancel()
|
||
key := workerTestID(t)
|
||
created, err := api.CreateWorkflowDefinition(ctx, key, CreateWorkflowDefinitionRequest{Name: "workflow." + key, Description: "精确数字发布回归"})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
id := created.Definition.ID
|
||
typ := WorkflowType{Kind: "OBJECT", Fields: map[string]WorkflowType{"n": {Kind: "NUMBER"}}}
|
||
task := WorkflowTask{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", InputType: typ, Input: WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`{"n":9007199254740993}`)}, ResultType: typ, TimeoutMS: 5000, Retry: RetryPolicy{MaxAttempts: 1, AttemptTimeoutMS: 3000}, RecoverySafety: RecoveryReconcileOnly}
|
||
in := PublishWorkflowVersionRequest{Version: "v1", Reason: "SDK publication", Specification: WorkflowSpecification{SchemaVersion: 1, TimeoutMS: 30000, FailurePolicy: "REVIEW", InputType: typ, Body: WorkflowBlock{Steps: []WorkflowStep{{Key: "prepare", Kind: "TASK", Task: &task}, {Key: "wait", Kind: "TIMER", Timer: &WorkflowTimer{DelayMS: 1000}}}, OutputType: typ, Output: WorkflowRef("prepare")}}}
|
||
// 真服务确认发布提交后，代理关闭客户端连接；SDK不得自动换键或默默重试。
|
||
target, _ := url.Parse(api.baseURL)
|
||
proxy := httputil.NewSingleHostReverseProxy(target)
|
||
var dropped atomic.Bool
|
||
var calls atomic.Int32
|
||
proxy.ModifyResponse = func(r *http.Response) error {
|
||
calls.Add(1)
|
||
if dropped.CompareAndSwap(false, true) {
|
||
raw, err := io.ReadAll(r.Body)
|
||
r.Body.Close()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if r.StatusCode != 201 || !strings.Contains(string(raw), `"deduplicated":false`) {
|
||
t.Error("publication was not committed before response loss")
|
||
}
|
||
return io.ErrUnexpectedEOF
|
||
}
|
||
return nil
|
||
}
|
||
proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
|
||
conn, _, e := w.(http.Hijacker).Hijack()
|
||
if e == nil {
|
||
conn.Close()
|
||
}
|
||
}
|
||
gateway := httptest.NewServer(proxy)
|
||
defer gateway.Close()
|
||
client, err := NewClient(gateway.URL, Options{Token: api.token})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := client.PublishWorkflowVersion(ctx, id, key, in); err == nil || calls.Load() != 1 || !dropped.Load() {
|
||
t.Fatal("response loss not surfaced", err)
|
||
}
|
||
published, err := client.PublishWorkflowVersion(ctx, id, key, in)
|
||
if err != nil || !published.Deduplicated {
|
||
t.Fatal("lost publication not recovered", err)
|
||
}
|
||
original := published.Version
|
||
if string(original.Specification.Body.Steps[0].Task.Input.Literal) != `{"n":9007199254740993}` || len(original.ContentDigest) != 64 || original.Analysis.Steps != 2 || original.Analysis.MaxTaskAttempts != 1 {
|
||
t.Fatal("SDK result changed precision or analysis")
|
||
}
|
||
same, err := api.PublishWorkflowVersion(ctx, id, workerTestID(t), in)
|
||
if err != nil || !same.Deduplicated || !reflect.DeepEqual(same.Version, original) {
|
||
t.Fatal("same version", err)
|
||
}
|
||
in.Specification.Body.Steps[0].Task.Input.Literal = json.RawMessage(`{"n":9007199254740992}`)
|
||
_, err = api.PublishWorkflowVersion(ctx, id, workerTestID(t), in)
|
||
var apiErr *APIError
|
||
if !errors.As(err, &apiErr) || apiErr.Code != "STATE_CONFLICT" {
|
||
t.Fatal("overwrite accepted", err)
|
||
}
|
||
in.Version = "v2"
|
||
next, err := api.PublishWorkflowVersion(ctx, id, workerTestID(t), in)
|
||
if err != nil || next.Version.ContentDigest == original.ContentDigest {
|
||
t.Fatal("v2", err)
|
||
}
|
||
stored, err := api.GetWorkflowVersion(ctx, id, "v1")
|
||
if err != nil || !reflect.DeepEqual(stored, original) {
|
||
t.Fatal("immutable v1", err)
|
||
}
|
||
metadata, err := api.GetWorkflowDefinition(ctx, id)
|
||
if err != nil || metadata.Revision != 3 {
|
||
t.Fatal("unexpected version count", err)
|
||
}
|
||
// 类型的空块输出支持显式null，不放宽普通任务的null参数约束。
|
||
in.Version = "v3"
|
||
in.Specification.Body = WorkflowBlock{OutputType: WorkflowType{Kind: "NULL"}, Output: WorkflowValue{Kind: "LITERAL", Literal: json.RawMessage(`null`)}}
|
||
v3, err := api.PublishWorkflowVersion(ctx, id, workerTestID(t), in)
|
||
if err != nil || v3.Version.Analysis.Queues == nil || len(v3.Version.Analysis.Queues) != 0 {
|
||
t.Fatal("null block", err)
|
||
}
|
||
if _, err := api.GetWorkflowVersion(ctx, id, "../other"); err == nil {
|
||
t.Fatal("unsafe path accepted")
|
||
}
|
||
t.Log("real HTTP publication response dropped after commit; original identity restored precise immutable version, no second revision; all SDK definition routes exercised")
|
||
}
|