mirror of https://github.com/dtm-labs/dtm.git
csharpjavadistributed-transactionsdtmgogolangmicroservicenodejsphpdatabasesagaseatatcctransactiontransactionsxapythondistributed
You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
279 lines
9.8 KiB
279 lines
9.8 KiB
/*
|
|
* Copyright (c) 2021 yedf. All rights reserved.
|
|
* Use of this source code is governed by a BSD-style
|
|
* license that can be found in the LICENSE file.
|
|
*/
|
|
|
|
package test
|
|
|
|
import (
|
|
"fmt"
|
|
"net/http"
|
|
"strconv"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/dtm-labs/dtm/client/dtmcli"
|
|
"github.com/dtm-labs/dtm/client/dtmcli/dtmimp"
|
|
"github.com/dtm-labs/dtm/dtmsvr/storage/registry"
|
|
"github.com/dtm-labs/dtm/dtmutil"
|
|
"github.com/dtm-labs/dtm/test/busi"
|
|
"github.com/stretchr/testify/assert"
|
|
)
|
|
|
|
func TestAPIVersion(t *testing.T) {
|
|
resp, err := dtmcli.GetRestyClient().R().Get(dtmutil.DefaultHTTPServer + "/version")
|
|
assert.Nil(t, err)
|
|
assert.Equal(t, 200, resp.StatusCode())
|
|
}
|
|
|
|
func TestAPIQuery(t *testing.T) {
|
|
gid := dtmimp.GetFuncName()
|
|
err := genMsg(gid).Submit()
|
|
assert.Nil(t, err)
|
|
waitTransProcessed(gid)
|
|
resp, err := dtmcli.GetRestyClient().R().SetQueryParam("gid", gid).Get(dtmutil.DefaultHTTPServer + "/query")
|
|
assert.Nil(t, err)
|
|
m := map[string]interface{}{}
|
|
assert.Equal(t, resp.StatusCode(), 200)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
assert.NotEqual(t, nil, m["transaction"])
|
|
assert.Equal(t, 2, len(m["branches"].([]interface{})))
|
|
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParam("gid", "").Get(dtmutil.DefaultHTTPServer + "/query")
|
|
e2p(err)
|
|
assert.Equal(t, resp.StatusCode(), 500)
|
|
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParam("gid", "1").Get(dtmutil.DefaultHTTPServer + "/query")
|
|
e2p(err)
|
|
assert.Equal(t, resp.StatusCode(), 200)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
assert.Equal(t, nil, m["transaction"])
|
|
assert.Equal(t, 0, len(m["branches"].([]interface{})))
|
|
}
|
|
|
|
func TestAPIAll(t *testing.T) {
|
|
startTime := time.Now()
|
|
for i := 0; i < 3; i++ { // add three
|
|
gid := dtmimp.GetFuncName() + fmt.Sprintf("%d", i)
|
|
err := genMsg(gid).Submit()
|
|
assert.Nil(t, err)
|
|
waitTransProcessed(gid)
|
|
}
|
|
endTime := time.Now()
|
|
|
|
resp, err := dtmcli.GetRestyClient().R().SetQueryParam("limit", "1").Get(dtmutil.DefaultHTTPServer + "/all")
|
|
assert.Nil(t, err)
|
|
m := map[string]interface{}{}
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos := m["next_position"].(string)
|
|
assert.NotEqual(t, "", nextPos)
|
|
// assert.Equal(t, 1, len(m["transactions"].([]interface{})))
|
|
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParam("gid", dtmimp.GetFuncName()+"1").Get(dtmutil.DefaultHTTPServer + "/all")
|
|
assert.Nil(t, err)
|
|
m = map[string]interface{}{}
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
assert.Equal(t, 1, len(m["transactions"].([]interface{})))
|
|
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"limit": "1",
|
|
"position": nextPos,
|
|
}).Get(dtmutil.DefaultHTTPServer + "/all")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos2 := m["next_position"].(string)
|
|
assert.NotEqual(t, "", nextPos2)
|
|
assert.NotEqual(t, nextPos, nextPos2)
|
|
// assert.Equal(t, 1, len(m["transactions"].([]interface{})))
|
|
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"limit": "1000",
|
|
"position": nextPos,
|
|
}).Get(dtmutil.DefaultHTTPServer + "/all")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos3 := m["next_position"].(string)
|
|
assert.Equal(t, "", nextPos3)
|
|
// assert.Equal(t, 2, len(m["transactions"].([]interface{}))) // the left 2.
|
|
|
|
// filter test
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"limit": "10",
|
|
"status": "succeed",
|
|
"transType": "msg",
|
|
"createTimeStart": strconv.Itoa(int(startTime.Add(time.Minute*-1).Unix() * 1000)),
|
|
"createTimeEnd": strconv.Itoa(int(endTime.Add(time.Minute*1).Unix() * 1000)),
|
|
}).Get(dtmutil.DefaultHTTPServer + "/all")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos1 := m["next_position"].(string)
|
|
// assert.Equal(t, 3, len(m["transactions"].([]interface{})))
|
|
assert.GreaterOrEqual(t, len(m["transactions"].([]interface{})), 3) // Be disturbed by something else test case, so use >=3 instead of =3.
|
|
assert.Empty(t, nextPos1) // is over
|
|
for _, item := range m["transactions"].([]interface{}) {
|
|
g := item.(map[string]interface{})
|
|
assert.Equal(t, "msg", g["trans_type"])
|
|
assert.Equal(t, "succeed", g["status"])
|
|
}
|
|
|
|
// filter, five minutes ago
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"limit": "10",
|
|
"status": "succeed",
|
|
"transType": "msg",
|
|
"createTimeStart": strconv.Itoa(int(startTime.Add(time.Minute*-10).Unix() * 1000)),
|
|
"createTimeEnd": strconv.Itoa(int(endTime.Add(time.Minute*-5).Unix() * 1000)),
|
|
}).Get(dtmutil.DefaultHTTPServer + "/all")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos1 = m["next_position"].(string)
|
|
assert.Equal(t, 0, len(m["transactions"].([]interface{})))
|
|
assert.Empty(t, nextPos1) // is over
|
|
|
|
//fmt.Printf("pos1:%s,pos2:%s,pos3:%s", nextPos, nextPos2, nextPos3)
|
|
}
|
|
|
|
func TestAPIScanKV(t *testing.T) {
|
|
for i := 0; i < 3; i++ { // add three
|
|
assert.Nil(t, httpSubscribe("test_topic"+fmt.Sprintf("%d", i), "http://dtm/test1"))
|
|
}
|
|
resp, err := dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"cat": "topics",
|
|
"limit": "1",
|
|
}).Get(dtmutil.DefaultHTTPServer + "/scanKV")
|
|
assert.Nil(t, err)
|
|
m := map[string]interface{}{}
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos := m["next_position"].(string)
|
|
assert.NotEqual(t, "", nextPos)
|
|
// assert.Equal(t, 1, len(m["kv"].([]interface{})))
|
|
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"cat": "topics",
|
|
"limit": "1",
|
|
"position": nextPos,
|
|
}).Get(dtmutil.DefaultHTTPServer + "/scanKV")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos2 := m["next_position"].(string)
|
|
assert.NotEqual(t, "", nextPos2)
|
|
assert.NotEqual(t, nextPos, nextPos2)
|
|
// assert.Equal(t, 1, len(m["kv"].([]interface{})))
|
|
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"cat": "topics",
|
|
"limit": "1000",
|
|
"position": nextPos,
|
|
}).Get(dtmutil.DefaultHTTPServer + "/scanKV")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
nextPos3 := m["next_position"].(string)
|
|
assert.Equal(t, "", nextPos3)
|
|
// assert.Equal(t, 2, len(m["kv"].([]interface{}))) // the left 2.
|
|
}
|
|
|
|
func TestAPIQueryKV(t *testing.T) {
|
|
m := map[string]interface{}{}
|
|
// normal
|
|
assert.Nil(t, httpSubscribe("test_topic_TestAPIQueryKV", "http://dtm/test1"))
|
|
resp, err := dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"cat": "topics",
|
|
"key": "test_topic_TestAPIQueryKV",
|
|
}).Get(dtmutil.DefaultHTTPServer + "/queryKV")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
assert.Equal(t, 1, len(m["kv"].([]interface{})))
|
|
|
|
// query non_existent topic
|
|
resp, err = dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"cat": "topics",
|
|
"key": "non_existent_topic_TestAPIQueryKV",
|
|
}).Get(dtmutil.DefaultHTTPServer + "/queryKV")
|
|
assert.Nil(t, err)
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
assert.Equal(t, 0, len(m["kv"].([]interface{})))
|
|
}
|
|
|
|
func TestDtmMetrics(t *testing.T) {
|
|
rest, err := dtmcli.GetRestyClient().R().Get("http://localhost:36789/api/metrics")
|
|
assert.Nil(t, err)
|
|
assert.Equal(t, rest.StatusCode(), 200)
|
|
}
|
|
|
|
func TestAPIResetCronTime(t *testing.T) {
|
|
testStoreResetCronTime(t, dtmimp.GetFuncName(), func(timeout int64, limit int64) (int64, bool, error) {
|
|
sTimeout := strconv.FormatInt(timeout, 10)
|
|
sLimit := strconv.FormatInt(limit, 10)
|
|
|
|
resp, err := dtmcli.GetRestyClient().R().SetQueryParams(map[string]string{
|
|
"timeout": sTimeout,
|
|
"limit": sLimit,
|
|
}).Get(dtmutil.DefaultHTTPServer + "/resetCronTime")
|
|
|
|
m := map[string]interface{}{}
|
|
dtmimp.MustUnmarshalString(resp.String(), &m)
|
|
hasRemaining, ok := m["has_remaining"].(bool)
|
|
assert.Equal(t, ok, true)
|
|
succeedCount, ok := m["succeed_count"].(float64)
|
|
assert.Equal(t, ok, true)
|
|
return int64(succeedCount), hasRemaining, err
|
|
})
|
|
}
|
|
|
|
func TestAPIForceStoppedNormal(t *testing.T) {
|
|
saga := genSaga(dtmimp.GetFuncName(), false, false)
|
|
busi.MainSwitch.TransOutResult.SetOnce("ONGOING")
|
|
saga.Submit()
|
|
waitTransProcessed(saga.Gid)
|
|
assert.Equal(t, StatusSubmitted, getTransStatus(saga.Gid))
|
|
|
|
resp, err := dtmcli.GetRestyClient().R().SetBody(map[string]string{
|
|
"gid": saga.Gid,
|
|
}).Post(dtmutil.DefaultHTTPServer + "/forceStop")
|
|
assert.Nil(t, err)
|
|
assert.Equal(t, resp.StatusCode(), http.StatusOK)
|
|
assert.Equal(t, StatusFailed, getTransStatus(saga.Gid))
|
|
}
|
|
|
|
func TestAPIForceStoppedAbnormal(t *testing.T) {
|
|
saga := genSaga(dtmimp.GetFuncName(), false, false)
|
|
saga.Submit()
|
|
waitTransProcessed(saga.Gid)
|
|
assert.Equal(t, []string{StatusPrepared, StatusSucceed, StatusPrepared, StatusSucceed}, getBranchesStatus(saga.Gid))
|
|
assert.Equal(t, StatusSucceed, getTransStatus(saga.Gid))
|
|
|
|
resp, err := dtmcli.GetRestyClient().R().SetBody(map[string]string{
|
|
"gid": saga.Gid,
|
|
}).Post(dtmutil.DefaultHTTPServer + "/forceStop")
|
|
assert.Nil(t, err)
|
|
assert.Equal(t, resp.StatusCode(), http.StatusConflict)
|
|
}
|
|
|
|
func TestAPIResetNextCronTime(t *testing.T) {
|
|
saga := genSaga(dtmimp.GetFuncName(), false, false)
|
|
saga.Submit()
|
|
waitTransProcessed(saga.Gid)
|
|
assert.Equal(t, []string{StatusPrepared, StatusSucceed, StatusPrepared, StatusSucceed}, getBranchesStatus(saga.Gid))
|
|
assert.Equal(t, StatusSucceed, getTransStatus(saga.Gid))
|
|
gid := saga.Gid
|
|
|
|
s := registry.GetStore()
|
|
g := s.FindTransGlobalStore(saga.Gid)
|
|
|
|
// reset
|
|
resp, err := dtmcli.GetRestyClient().R().SetBody(map[string]string{
|
|
"gid": saga.Gid,
|
|
}).Post(dtmutil.DefaultHTTPServer + "/resetNextCronTime")
|
|
assert.Nil(t, err)
|
|
assert.Equal(t, resp.StatusCode(), http.StatusOK)
|
|
|
|
// after reset assert
|
|
g2 := s.FindTransGlobalStore(gid)
|
|
assert.NotNil(t, g2)
|
|
assert.Equal(t, gid, g2.Gid)
|
|
assert.Greater(t, time.Now().Add(3*time.Second), *g2.NextCronTime)
|
|
assert.Equal(t, g2.UpdateTime, g2.NextCronTime)
|
|
assert.NotEqual(t, g.UpdateTime, g2.UpdateTime)
|
|
assert.NotEqual(t, g.NextCronTime, g2.NextCronTime)
|
|
}
|
|
|