sink_node_test.go 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  1. // Copyright 2021 EMQ Technologies Co., Ltd.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package node
  15. import (
  16. "fmt"
  17. "github.com/lf-edge/ekuiper/internal/conf"
  18. "github.com/lf-edge/ekuiper/internal/topo/context"
  19. "github.com/lf-edge/ekuiper/internal/topo/topotest/mocknode"
  20. "reflect"
  21. "testing"
  22. "time"
  23. )
  24. func TestSinkTemplate_Apply(t *testing.T) {
  25. conf.InitConf()
  26. var tests = []struct {
  27. config map[string]interface{}
  28. data []byte
  29. result [][]byte
  30. }{
  31. {
  32. config: map[string]interface{}{
  33. "sendSingle": true,
  34. "dataTemplate": `{"wrapper":"w1","content":{{toJson .}},"ab":"{{.ab}}"}`,
  35. },
  36. data: []byte(`[{"ab":"hello1"},{"ab":"hello2"}]`),
  37. result: [][]byte{[]byte(`{"wrapper":"w1","content":{"ab":"hello1"},"ab":"hello1"}`), []byte(`{"wrapper":"w1","content":{"ab":"hello2"},"ab":"hello2"}`)},
  38. }, {
  39. config: map[string]interface{}{
  40. "dataTemplate": `{"wrapper":"arr","content":{{json .}},"content0":{{json (index . 0)}},ab0":"{{index . 0 "ab"}}"}`,
  41. },
  42. data: []byte(`[{"ab":"hello1"},{"ab":"hello2"}]`),
  43. result: [][]byte{[]byte(`{"wrapper":"arr","content":[{"ab":"hello1"},{"ab":"hello2"}],"content0":{"ab":"hello1"},ab0":"hello1"}`)},
  44. }, {
  45. config: map[string]interface{}{
  46. "dataTemplate": `<div>results</div><ul>{{range .}}<li>{{.ab}}</li>{{end}}</ul>`,
  47. },
  48. data: []byte(`[{"ab":"hello1"},{"ab":"hello2"}]`),
  49. result: [][]byte{[]byte(`<div>results</div><ul><li>hello1</li><li>hello2</li></ul>`)},
  50. }, {
  51. config: map[string]interface{}{
  52. "dataTemplate": `{"content":{{toJson .}}}`,
  53. },
  54. data: []byte(`[{"ab":"hello1"},{"ab":"hello2"}]`),
  55. result: [][]byte{[]byte(`{"content":[{"ab":"hello1"},{"ab":"hello2"}]}`)},
  56. }, {
  57. config: map[string]interface{}{
  58. "sendSingle": true,
  59. "dataTemplate": `{"newab":"{{.ab}}"}`,
  60. },
  61. data: []byte(`[{"ab":"hello1"},{"ab":"hello2"}]`),
  62. result: [][]byte{[]byte(`{"newab":"hello1"}`), []byte(`{"newab":"hello2"}`)},
  63. }, {
  64. config: map[string]interface{}{
  65. "sendSingle": true,
  66. "dataTemplate": `{"newab":"{{.ab}}"}`,
  67. },
  68. data: []byte(`[{"ab":"hello1"},{"ab":"hello2"}]`),
  69. result: [][]byte{[]byte(`{"newab":"hello1"}`), []byte(`{"newab":"hello2"}`)},
  70. }, {
  71. config: map[string]interface{}{
  72. "sendSingle": true,
  73. "dataTemplate": `{"__meta":{{toJson .__meta}},"temp":{{.temperature}}}`,
  74. },
  75. data: []byte(`[{"temperature":33,"humidity":70,"__meta": {"messageid":45,"other": "mock"}}]`),
  76. result: [][]byte{[]byte(`{"__meta":{"messageid":45,"other":"mock"},"temp":33}`)},
  77. }, {
  78. config: map[string]interface{}{
  79. "dataTemplate": `[{"__meta":{{toJson (index . 0 "__meta")}},"temp":{{index . 0 "temperature"}}}]`,
  80. },
  81. data: []byte(`[{"temperature":33,"humidity":70,"__meta": {"messageid":45,"other": "mock"}}]`),
  82. result: [][]byte{[]byte(`[{"__meta":{"messageid":45,"other":"mock"},"temp":33}]`)},
  83. }, {
  84. config: map[string]interface{}{
  85. "dataTemplate": `[{{range $index, $ele := .}}{{if $index}},{{end}}{"result":{{add $ele.temperature $ele.humidity}}}{{end}}]`,
  86. },
  87. data: []byte(`[{"temperature":33,"humidity":70},{"temperature":22.0,"humidity":50},{"temperature":11,"humidity":90}]`),
  88. result: [][]byte{[]byte(`[{"result":103},{"result":72},{"result":101}]`)},
  89. }, {
  90. config: map[string]interface{}{
  91. "dataTemplate": `{{$counter := 0}}{{range $index, $ele := .}}{{if ne 90.0 $ele.humidity}}{{$counter = add $counter 1}}{{end}}{{end}}{"result":{{$counter}}}`,
  92. },
  93. data: []byte(`[{"temperature":33,"humidity":70},{"temperature":22,"humidity":50},{"temperature":11,"humidity":90}]`),
  94. result: [][]byte{[]byte(`{"result":2}`)},
  95. }, {
  96. config: map[string]interface{}{
  97. "dataTemplate": `{"a":"{{base64 .a}}","b":"{{base64 .b}}","c":"{{b64enc .c}}","d":"{{b64enc .d}}","e":"{{base64 .e}}"}`,
  98. "sendSingle": true,
  99. },
  100. data: []byte(`[{"a":1,"b":3.1415,"c":"hello","d":"{\"hello\" : 3}","e":{"humidity":20,"temperature":30}}]`),
  101. result: [][]byte{[]byte(`{"a":"MQ==","b":"My4xNDE1","c":"aGVsbG8=","d":"eyJoZWxsbyIgOiAzfQ==","e":"eyJodW1pZGl0eSI6MjAsInRlbXBlcmF0dXJlIjozMH0="}`)},
  102. },
  103. }
  104. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  105. contextLogger := conf.Log.WithField("rule", "TestSinkTemplate_Apply")
  106. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  107. for i, tt := range tests {
  108. mockSink := mocknode.NewMockSink()
  109. s := NewSinkNodeWithSink("mockSink", mockSink, tt.config)
  110. s.Open(ctx, make(chan error))
  111. s.input <- tt.data
  112. time.Sleep(1 * time.Second)
  113. s.close(ctx, contextLogger)
  114. results := mockSink.GetResults()
  115. if !reflect.DeepEqual(tt.result, results) {
  116. t.Errorf("%d \tresult mismatch:\n\nexp=%s\n\ngot=%s\n\n", i, tt.result, results)
  117. }
  118. }
  119. }