status-im
/
consul
mirror of https://github.com/status-im/consul.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity
consul/agent/txn_endpoint_test.go

605 lines
14 KiB
Go
Raw Blame History

package agent
import (
"bytes"
"encoding/base64"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
"github.com/hashicorp/consul/api"
"github.com/hashicorp/consul/testrpc"
"github.com/hashicorp/raft"
"github.com/pascaldekloe/goe/verify"
"github.com/hashicorp/consul/agent/structs"
)
func TestTxnEndpoint_Bad_JSON(t *testing.T) {
t.Parallel()
a := NewTestAgent(t, t.Name(), "")
defer a.Shutdown()
buf := bytes.NewBuffer([]byte("{"))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
if _, err := a.srv.Txn(resp, req); err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 400 {
t.Fatalf("expected 400, got %d", resp.Code)
}
if !bytes.Contains(resp.Body.Bytes(), []byte("Failed to parse")) {
t.Fatalf("expected conflicting args error")
}
}
func TestTxnEndpoint_Bad_Size_Item(t *testing.T) {
t.Parallel()
testIt := func(agent *TestAgent, wantPass bool) {
value := strings.Repeat("X", 3*raft.SuggestedMaxDataSize)
value = base64.StdEncoding.EncodeToString([]byte(value))
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "set",
"Key": "key",
"Value": %q
}
}
]
`, value)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
if _, err := agent.srv.Txn(resp, req); err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 413 && !wantPass {
t.Fatalf("expected 413, got %d", resp.Code)
}
if resp.Code != 200 && wantPass {
t.Fatalf("expected 200, got %d", resp.Code)
}
}
t.Run("toobig", func(t *testing.T) {
a := NewTestAgent(t, t.Name(), "")
testIt(a, false)
a.Shutdown()
})
t.Run("allowed", func(t *testing.T) {
a := NewTestAgent(t, t.Name(), "limits = { kv_max_value_size = 123456789 }")
testIt(a, true)
a.Shutdown()
})
}
func TestTxnEndpoint_Bad_Size_Net(t *testing.T) {
t.Parallel()
testIt := func(agent *TestAgent, wantPass bool) {
value := strings.Repeat("X", 3*raft.SuggestedMaxDataSize)
value = base64.StdEncoding.EncodeToString([]byte(value))
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "set",
"Key": "key1",
"Value": %q
}
},
{
"KV": {
"Verb": "set",
"Key": "key1",
"Value": %q
}
},
{
"KV": {
"Verb": "set",
"Key": "key1",
"Value": %q
}
}
]
`, value, value, value)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
if _, err := agent.srv.Txn(resp, req); err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 413 && !wantPass {
t.Fatalf("expected 413, got %d", resp.Code)
}
if resp.Code != 200 && wantPass {
t.Fatalf("expected 200, got %d", resp.Code)
}
}
t.Run("toobig", func(t *testing.T) {
a := NewTestAgent(t, t.Name(), "")
testIt(a, false)
a.Shutdown()
})
t.Run("allowed", func(t *testing.T) {
a := NewTestAgent(t, t.Name(), "limits = { kv_max_value_size = 123456789 }")
testIt(a, true)
a.Shutdown()
})
}
func TestTxnEndpoint_Bad_Size_Ops(t *testing.T) {
t.Parallel()
a := NewTestAgent(t, t.Name(), "")
defer a.Shutdown()
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
%s
{
"KV": {
"Verb": "set",
"Key": "key",
"Value": ""
}
}
]
`, strings.Repeat(`{ "KV": { "Verb": "get", "Key": "key" } },`, 2*maxTxnOps))))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
if _, err := a.srv.Txn(resp, req); err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 413 {
t.Fatalf("expected 413, got %d", resp.Code)
}
}
func TestTxnEndpoint_KV_Actions(t *testing.T) {
t.Parallel()
t.Run("", func(t *testing.T) {
a := NewTestAgent(t, t.Name(), "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
// Make sure all incoming fields get converted properly to the internal
// RPC format.
var index uint64
id := makeTestSession(t, a.srv)
{
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "lock",
"Key": "key",
"Value": "aGVsbG8gd29ybGQ=",
"Flags": 23,
"Session": %q
}
},
{
"KV": {
"Verb": "get",
"Key": "key"
}
}
]
`, id)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
if err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 200 {
t.Fatalf("expected 200, got %d", resp.Code)
}
txnResp, ok := obj.(structs.TxnResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
if len(txnResp.Results) != 2 {
t.Fatalf("bad: %v", txnResp)
}
index = txnResp.Results[0].KV.ModifyIndex
expected := structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: nil,
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
},
},
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("hello world"),
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
},
},
},
}
if !reflect.DeepEqual(txnResp, expected) {
t.Fatalf("bad: %v", txnResp)
}
}
// Do a read-only transaction that should get routed to the
// fast-path endpoint.
{
buf := bytes.NewBuffer([]byte(`
[
{
"KV": {
"Verb": "get",
"Key": "key"
}
},
{
"KV": {
"Verb": "get-tree",
"Key": "key"
}
}
]
`))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
if err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 200 {
t.Fatalf("expected 200, got %d", resp.Code)
}
header := resp.Header().Get("X-Consul-KnownLeader")
if header != "true" {
t.Fatalf("bad: %v", header)
}
header = resp.Header().Get("X-Consul-LastContact")
if header != "0" {
t.Fatalf("bad: %v", header)
}
txnResp, ok := obj.(structs.TxnReadResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
expected := structs.TxnReadResponse{
TxnResponse: structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("hello world"),
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
},
},
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("hello world"),
Flags: 23,
Session: id,
LockIndex: 1,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
},
},
},
},
QueryMeta: structs.QueryMeta{
KnownLeader: true,
},
}
if !reflect.DeepEqual(txnResp, expected) {
t.Fatalf("bad: %v", txnResp)
}
}
// Now that we have an index we can do a CAS to make sure the
// index field gets translated to the RPC format.
{
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"KV": {
"Verb": "cas",
"Key": "key",
"Value": "Z29vZGJ5ZSB3b3JsZA==",
"Index": %d
}
},
{
"KV": {
"Verb": "get",
"Key": "key"
}
}
]
`, index)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
if err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 200 {
t.Fatalf("expected 200, got %d", resp.Code)
}
txnResp, ok := obj.(structs.TxnResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
if len(txnResp.Results) != 2 {
t.Fatalf("bad: %v", txnResp)
}
modIndex := txnResp.Results[0].KV.ModifyIndex
expected := structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: nil,
Session: id,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: modIndex,
},
},
},
&structs.TxnResult{
KV: &structs.DirEntry{
Key: "key",
Value: []byte("goodbye world"),
Session: id,
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: modIndex,
},
},
},
},
}
if !reflect.DeepEqual(txnResp, expected) {
t.Fatalf("bad: %v", txnResp)
}
}
})
// Verify an error inside a transaction.
t.Run("", func(t *testing.T) {
a := NewTestAgent(t, t.Name(), "")
defer a.Shutdown()
buf := bytes.NewBuffer([]byte(`
[
{
"KV": {
"Verb": "lock",
"Key": "key",
"Value": "aGVsbG8gd29ybGQ=",
"Session": "nope"
}
},
{
"KV": {
"Verb": "get",
"Key": "key"
}
}
]
`))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
if _, err := a.srv.Txn(resp, req); err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 409 {
t.Fatalf("expected 409, got %d", resp.Code)
}
if !bytes.Contains(resp.Body.Bytes(), []byte("failed session lookup")) {
t.Fatalf("bad: %s", resp.Body.String())
}
})
}
func TestTxnEndpoint_UpdateCheck(t *testing.T) {
t.Parallel()
a := NewTestAgent(t, t.Name(), "")
defer a.Shutdown()
testrpc.WaitForTestAgent(t, a.RPC, "dc1")
// Make sure the fields of a check are handled correctly when both creating and
// updating, and test both sets of duration fields to ensure backwards compatibility.
buf := bytes.NewBuffer([]byte(fmt.Sprintf(`
[
{
"Check": {
"Verb": "set",
"Check": {
"Node": "%s",
"CheckID": "nodecheck",
"Name": "Node http check",
"Status": "critical",
"Notes": "Http based health check",
"Output": "",
"ServiceID": "",
"ServiceName": "",
"Definition": {
"Interval": "6s",
"Timeout": "6s",
"DeregisterCriticalServiceAfter": "6s",
"HTTP": "http://localhost:8000",
"TLSSkipVerify": true
}
}
}
},
{
"Check": {
"Verb": "set",
"Check": {
"Node": "%s",
"CheckID": "nodecheck",
"Name": "Node http check",
"Status": "passing",
"Notes": "Http based health check",
"Output": "success",
"ServiceID": "",
"ServiceName": "",
"Definition": {
"Interval": "10s",
"Timeout": "10s",
"DeregisterCriticalServiceAfter": "15m",
"HTTP": "http://localhost:9000",
"TLSSkipVerify": false
}
}
}
},
{
"Check": {
"Verb": "set",
"Check": {
"Node": "%s",
"CheckID": "nodecheck",
"Name": "Node http check",
"Status": "passing",
"Notes": "Http based health check",
"Output": "success",
"ServiceID": "",
"ServiceName": "",
"Definition": {
"IntervalDuration": "15s",
"TimeoutDuration": "15s",
"DeregisterCriticalServiceAfterDuration": "30m",
"HTTP": "http://localhost:9000",
"TLSSkipVerify": false
}
}
}
}
]
`, a.config.NodeName, a.config.NodeName, a.config.NodeName)))
req, _ := http.NewRequest("PUT", "/v1/txn", buf)
resp := httptest.NewRecorder()
obj, err := a.srv.Txn(resp, req)
if err != nil {
t.Fatalf("err: %v", err)
}
if resp.Code != 200 {
t.Fatalf("expected 200, got %d", resp.Code)
}
txnResp, ok := obj.(structs.TxnResponse)
if !ok {
t.Fatalf("bad type: %T", obj)
}
if len(txnResp.Results) != 3 {
t.Fatalf("bad: %v", txnResp)
}
index := txnResp.Results[0].Check.ModifyIndex
expected := structs.TxnResponse{
Results: structs.TxnResults{
&structs.TxnResult{
Check: &structs.HealthCheck{
Node: a.config.NodeName,
CheckID: "nodecheck",
Name: "Node http check",
Status: api.HealthCritical,
Notes: "Http based health check",
Definition: structs.HealthCheckDefinition{
Interval: 6 * time.Second,
Timeout: 6 * time.Second,
DeregisterCriticalServiceAfter: 6 * time.Second,
HTTP: "http://localhost:8000",
TLSSkipVerify: true,
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
},
},
&structs.TxnResult{
Check: &structs.HealthCheck{
Node: a.config.NodeName,
CheckID: "nodecheck",
Name: "Node http check",
Status: api.HealthPassing,
Notes: "Http based health check",
Output: "success",
Definition: structs.HealthCheckDefinition{
Interval: 10 * time.Second,
Timeout: 10 * time.Second,
DeregisterCriticalServiceAfter: 15 * time.Minute,
HTTP: "http://localhost:9000",
TLSSkipVerify: false,
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
},
},
&structs.TxnResult{
Check: &structs.HealthCheck{
Node: a.config.NodeName,
CheckID: "nodecheck",
Name: "Node http check",
Status: api.HealthPassing,
Notes: "Http based health check",
Output: "success",
Definition: structs.HealthCheckDefinition{
Interval: 15 * time.Second,
Timeout: 15 * time.Second,
DeregisterCriticalServiceAfter: 30 * time.Minute,
HTTP: "http://localhost:9000",
TLSSkipVerify: false,
},
RaftIndex: structs.RaftIndex{
CreateIndex: index,
ModifyIndex: index,
},
},
},
},
}
verify.Values(t, "", txnResp, expected)
}
Reference in New Issue View Git Blame Copy Permalink