misc_func_test.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864
  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 operator
  15. import (
  16. "encoding/json"
  17. "fmt"
  18. "github.com/lf-edge/ekuiper/internal/conf"
  19. "github.com/lf-edge/ekuiper/internal/testx"
  20. "github.com/lf-edge/ekuiper/internal/topo/context"
  21. "github.com/lf-edge/ekuiper/internal/xsql"
  22. "reflect"
  23. "strings"
  24. "testing"
  25. )
  26. func TestMiscFunc_Apply1(t *testing.T) {
  27. var tests = []struct {
  28. sql string
  29. data *xsql.Tuple
  30. result []map[string]interface{}
  31. }{
  32. {
  33. sql: "SELECT md5(a) AS a FROM test",
  34. data: &xsql.Tuple{
  35. Emitter: "test",
  36. Message: xsql.Message{
  37. "a": "The quick brown fox jumps over the lazy dog",
  38. "b": "myb",
  39. "c": "myc",
  40. },
  41. },
  42. result: []map[string]interface{}{{
  43. "a": strings.ToLower("9E107D9D372BB6826BD81D3542A419D6"),
  44. }},
  45. },
  46. {
  47. sql: "SELECT md5(d) AS a FROM test",
  48. data: &xsql.Tuple{
  49. Emitter: "test",
  50. Message: xsql.Message{
  51. "a": "The quick brown fox jumps over the lazy dog",
  52. "b": "myb",
  53. "c": "myc",
  54. },
  55. },
  56. result: []map[string]interface{}{{}},
  57. },
  58. {
  59. sql: "SELECT sha1(a) AS a FROM test",
  60. data: &xsql.Tuple{
  61. Emitter: "test",
  62. Message: xsql.Message{
  63. "a": "The quick brown fox jumps over the lazy dog",
  64. "b": "myb",
  65. "c": "myc",
  66. },
  67. },
  68. result: []map[string]interface{}{{
  69. "a": strings.ToLower("2FD4E1C67A2D28FCED849EE1BB76E7391B93EB12"),
  70. }},
  71. },
  72. {
  73. sql: "SELECT sha256(a) AS a FROM test",
  74. data: &xsql.Tuple{
  75. Emitter: "test",
  76. Message: xsql.Message{
  77. "a": "The quick brown fox jumps over the lazy dog",
  78. "b": "myb",
  79. "c": "myc",
  80. },
  81. },
  82. result: []map[string]interface{}{{
  83. "a": strings.ToLower("D7A8FBB307D7809469CA9ABCB0082E4F8D5651E46D3CDB762D02D0BF37C9E592"),
  84. }},
  85. },
  86. {
  87. sql: "SELECT sha384(a) AS a FROM test",
  88. data: &xsql.Tuple{
  89. Emitter: "test",
  90. Message: xsql.Message{
  91. "a": "The quick brown fox jumps over the lazy dog",
  92. "b": "myb",
  93. "c": "myc",
  94. },
  95. },
  96. result: []map[string]interface{}{{
  97. "a": strings.ToLower("CA737F1014A48F4C0B6DD43CB177B0AFD9E5169367544C494011E3317DBF9A509CB1E5DC1E85A941BBEE3D7F2AFBC9B1"),
  98. }},
  99. },
  100. {
  101. sql: "SELECT sha512(a) AS a FROM test",
  102. data: &xsql.Tuple{
  103. Emitter: "test",
  104. Message: xsql.Message{
  105. "a": "The quick brown fox jumps over the lazy dog",
  106. "b": "myb",
  107. "c": "myc",
  108. },
  109. },
  110. result: []map[string]interface{}{{
  111. "a": strings.ToLower("07E547D9586F6A73F73FBAC0435ED76951218FB7D0C8D788A309D785436BBB642E93A252A954F23912547D1E8A3B5ED6E1BFD7097821233FA0538F3DB854FEE6"),
  112. }},
  113. },
  114. {
  115. sql: "SELECT mqtt(topic) AS a FROM test",
  116. data: &xsql.Tuple{
  117. Emitter: "test",
  118. Message: xsql.Message{},
  119. Metadata: xsql.Metadata{
  120. "topic": "devices/device_001/message",
  121. },
  122. },
  123. result: []map[string]interface{}{{
  124. "a": "devices/device_001/message",
  125. }},
  126. },
  127. {
  128. sql: "SELECT mqtt(topic) AS a FROM test",
  129. data: &xsql.Tuple{
  130. Emitter: "test",
  131. Message: xsql.Message{},
  132. Metadata: xsql.Metadata{
  133. "topic": "devices/device_001/message",
  134. },
  135. },
  136. result: []map[string]interface{}{{
  137. "a": "devices/device_001/message",
  138. }},
  139. },
  140. {
  141. sql: "SELECT topic, mqtt(topic) AS a FROM test",
  142. data: &xsql.Tuple{
  143. Emitter: "test",
  144. Message: xsql.Message{
  145. "topic": "fff",
  146. },
  147. Metadata: xsql.Metadata{
  148. "topic": "devices/device_001/message",
  149. },
  150. },
  151. result: []map[string]interface{}{{
  152. "topic": "fff",
  153. "a": "devices/device_001/message",
  154. }},
  155. },
  156. {
  157. sql: "SELECT cardinality(arr) as r FROM test",
  158. data: &xsql.Tuple{
  159. Emitter: "test",
  160. Message: xsql.Message{
  161. "temperature": 43.2,
  162. "arr": []int{},
  163. },
  164. },
  165. result: []map[string]interface{}{{
  166. "r": float64(0),
  167. }},
  168. },
  169. {
  170. sql: "SELECT cardinality(arr) as r FROM test",
  171. data: &xsql.Tuple{
  172. Emitter: "test",
  173. Message: xsql.Message{
  174. "temperature": 43.2,
  175. "arr": []int{1, 2, 3, 4, 5},
  176. },
  177. },
  178. result: []map[string]interface{}{{
  179. "r": float64(5),
  180. }},
  181. },
  182. {
  183. sql: "SELECT isNull(arr) as r FROM test",
  184. data: &xsql.Tuple{
  185. Emitter: "test",
  186. Message: xsql.Message{
  187. "temperature": 43.2,
  188. "arr": []int{},
  189. },
  190. },
  191. result: []map[string]interface{}{{
  192. "r": false,
  193. }},
  194. },
  195. {
  196. sql: "SELECT isNull(arr) as r FROM test",
  197. data: &xsql.Tuple{
  198. Emitter: "test",
  199. Message: xsql.Message{
  200. "temperature": 43.2,
  201. "arr": []float64(nil),
  202. },
  203. },
  204. result: []map[string]interface{}{{
  205. "r": true,
  206. }},
  207. },
  208. {
  209. sql: "SELECT isNull(rec) as r FROM test",
  210. data: &xsql.Tuple{
  211. Emitter: "test",
  212. Message: xsql.Message{
  213. "temperature": 43.2,
  214. "rec": map[string]interface{}(nil),
  215. },
  216. },
  217. result: []map[string]interface{}{{
  218. "r": true,
  219. }},
  220. },
  221. {
  222. sql: "SELECT cast(a * 1000, \"datetime\") AS a FROM test",
  223. data: &xsql.Tuple{
  224. Emitter: "test",
  225. Message: xsql.Message{
  226. "a": 1.62000273e+09,
  227. "b": "ya",
  228. "c": "myc",
  229. },
  230. },
  231. result: []map[string]interface{}{{
  232. "a": "2021-05-03T00:45:30Z",
  233. }},
  234. },
  235. }
  236. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  237. contextLogger := conf.Log.WithField("rule", "TestMiscFunc_Apply1")
  238. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  239. for i, tt := range tests {
  240. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  241. if err != nil || stmt == nil {
  242. t.Errorf("parse sql %s error %v", tt.sql, err)
  243. }
  244. pp := &ProjectOp{Fields: stmt.Fields}
  245. fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters)
  246. result := pp.Apply(ctx, tt.data, fv, afv)
  247. var mapRes []map[string]interface{}
  248. if v, ok := result.([]byte); ok {
  249. err := json.Unmarshal(v, &mapRes)
  250. if err != nil {
  251. t.Errorf("Failed to parse the input into map.\n")
  252. continue
  253. }
  254. if !reflect.DeepEqual(tt.result, mapRes) {
  255. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  256. }
  257. } else {
  258. t.Errorf("The returned result is not type of []byte\n")
  259. }
  260. }
  261. }
  262. func TestMqttFunc_Apply2(t *testing.T) {
  263. var tests = []struct {
  264. sql string
  265. data *xsql.JoinTupleSets
  266. result []map[string]interface{}
  267. }{
  268. {
  269. sql: "SELECT id1, mqtt(src1.topic) AS a, mqtt(src2.topic) as b FROM src1 LEFT JOIN src2 ON src1.id1 = src2.id1",
  270. data: &xsql.JoinTupleSets{
  271. Content: []xsql.JoinTuple{
  272. {
  273. Tuples: []xsql.Tuple{
  274. {Emitter: "src1", Message: xsql.Message{"id1": "1", "f1": "v1"}, Metadata: xsql.Metadata{"topic": "devices/type1/device001"}},
  275. {Emitter: "src2", Message: xsql.Message{"id2": "1", "f2": "w1"}, Metadata: xsql.Metadata{"topic": "devices/type2/device001"}},
  276. },
  277. },
  278. },
  279. },
  280. result: []map[string]interface{}{{
  281. "id1": "1",
  282. "a": "devices/type1/device001",
  283. "b": "devices/type2/device001",
  284. }},
  285. },
  286. }
  287. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  288. contextLogger := conf.Log.WithField("rule", "TestMqttFunc_Apply2")
  289. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  290. for i, tt := range tests {
  291. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  292. if err != nil || stmt == nil {
  293. t.Errorf("parse sql %s error %v", tt.sql, err)
  294. }
  295. pp := &ProjectOp{Fields: stmt.Fields}
  296. fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters)
  297. result := pp.Apply(ctx, tt.data, fv, afv)
  298. var mapRes []map[string]interface{}
  299. if v, ok := result.([]byte); ok {
  300. err := json.Unmarshal(v, &mapRes)
  301. if err != nil {
  302. t.Errorf("Failed to parse the input into map.\n")
  303. continue
  304. }
  305. //fmt.Printf("%t\n", mapRes["kuiper_field_0"])
  306. if !reflect.DeepEqual(tt.result, mapRes) {
  307. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  308. }
  309. } else {
  310. t.Errorf("The returned result is not type of []byte\n")
  311. }
  312. }
  313. }
  314. func TestMetaFunc_Apply1(t *testing.T) {
  315. var tests = []struct {
  316. sql string
  317. data interface{}
  318. result interface{}
  319. }{
  320. {
  321. sql: "SELECT topic, meta(topic) AS a FROM test",
  322. data: &xsql.Tuple{
  323. Emitter: "test",
  324. Message: xsql.Message{
  325. "topic": "fff",
  326. },
  327. Metadata: xsql.Metadata{
  328. "topic": "devices/device_001/message",
  329. },
  330. },
  331. result: []map[string]interface{}{{
  332. "topic": "fff",
  333. "a": "devices/device_001/message",
  334. }},
  335. },
  336. {
  337. sql: "SELECT meta(device) as d, meta(temperature->device) as r FROM test",
  338. data: &xsql.Tuple{
  339. Emitter: "test",
  340. Message: xsql.Message{
  341. "temperature": 43.2,
  342. },
  343. Metadata: xsql.Metadata{
  344. "temperature": map[string]interface{}{
  345. "id": "dfadfasfas",
  346. "device": "device2",
  347. },
  348. "device": "gateway",
  349. },
  350. },
  351. result: []map[string]interface{}{{
  352. "d": "gateway",
  353. "r": "device2",
  354. }},
  355. },
  356. {
  357. sql: "SELECT meta(*) as r FROM test",
  358. data: &xsql.Tuple{
  359. Emitter: "test",
  360. Message: xsql.Message{
  361. "temperature": 43.2,
  362. },
  363. Metadata: xsql.Metadata{
  364. "temperature": map[string]interface{}{
  365. "id": "dfadfasfas",
  366. "device": "device2",
  367. },
  368. "device": "gateway",
  369. },
  370. },
  371. result: []map[string]interface{}{{
  372. "r": map[string]interface{}{
  373. "temperature": map[string]interface{}{
  374. "id": "dfadfasfas",
  375. "device": "device2",
  376. },
  377. "device": "gateway",
  378. },
  379. }},
  380. },
  381. {
  382. sql: "SELECT topic, meta(`Light-diming`->device) AS a FROM test",
  383. data: &xsql.Tuple{
  384. Emitter: "test",
  385. Message: xsql.Message{
  386. "topic": "fff",
  387. },
  388. Metadata: xsql.Metadata{
  389. "Light-diming": map[string]interface{}{
  390. "device": "device2",
  391. },
  392. },
  393. },
  394. result: []map[string]interface{}{{
  395. "topic": "fff",
  396. "a": "device2",
  397. }},
  398. },
  399. }
  400. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  401. contextLogger := conf.Log.WithField("rule", "TestMetaFunc_Apply1")
  402. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  403. for i, tt := range tests {
  404. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  405. if err != nil || stmt == nil {
  406. t.Errorf("parse sql %s error %v", tt.sql, err)
  407. }
  408. pp := &ProjectOp{Fields: stmt.Fields}
  409. fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters)
  410. result := pp.Apply(ctx, tt.data, fv, afv)
  411. var mapRes []map[string]interface{}
  412. if v, ok := result.([]byte); ok {
  413. err := json.Unmarshal(v, &mapRes)
  414. if err != nil {
  415. t.Errorf("Failed to parse the input into map.\n")
  416. continue
  417. }
  418. //fmt.Printf("%t\n", mapRes["kuiper_field_0"])
  419. if !reflect.DeepEqual(tt.result, mapRes) {
  420. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  421. }
  422. } else {
  423. t.Errorf("The returned result is not type of []byte\n")
  424. }
  425. }
  426. }
  427. func TestJsonPathFunc_Apply1(t *testing.T) {
  428. var tests = []struct {
  429. sql string
  430. data interface{}
  431. result interface{}
  432. err string
  433. }{
  434. {
  435. sql: `SELECT json_path_query(equipment, "$.arm_right") AS a FROM test`,
  436. data: &xsql.Tuple{
  437. Emitter: "test",
  438. Message: xsql.Message{
  439. "class": "warrior",
  440. "equipment": map[string]interface{}{
  441. "rings": []map[string]interface{}{
  442. {
  443. "name": "ring of despair",
  444. "weight": 0.1,
  445. }, {
  446. "name": "ring of strength",
  447. "weight": 2.4,
  448. },
  449. },
  450. "arm_right": "Sword of flame",
  451. "arm_left": "Shield of faith",
  452. },
  453. },
  454. },
  455. result: []map[string]interface{}{{
  456. "a": "Sword of flame",
  457. }},
  458. }, {
  459. sql: `SELECT json_path_query(equipment, "$.rings[*].weight") AS a FROM test`,
  460. data: &xsql.Tuple{
  461. Emitter: "test",
  462. Message: xsql.Message{
  463. "class": "warrior",
  464. "equipment": map[string]interface{}{
  465. "rings": []interface{}{
  466. map[string]interface{}{
  467. "name": "ring of despair",
  468. "weight": 0.1,
  469. }, map[string]interface{}{
  470. "name": "ring of strength",
  471. "weight": 2.4,
  472. },
  473. },
  474. "arm_right": "Sword of flame",
  475. "arm_left": "Shield of faith",
  476. },
  477. },
  478. },
  479. result: []map[string]interface{}{{
  480. "a": []interface{}{
  481. 0.1, 2.4,
  482. },
  483. }},
  484. }, {
  485. sql: `SELECT json_path_query_first(equipment, "$.rings[*].weight") AS a FROM test`,
  486. data: &xsql.Tuple{
  487. Emitter: "test",
  488. Message: xsql.Message{
  489. "class": "warrior",
  490. "equipment": map[string]interface{}{
  491. "rings": []interface{}{
  492. map[string]interface{}{
  493. "name": "ring of despair",
  494. "weight": 0.1,
  495. }, map[string]interface{}{
  496. "name": "ring of strength",
  497. "weight": 2.4,
  498. },
  499. },
  500. "arm_right": "Sword of flame",
  501. "arm_left": "Shield of faith",
  502. },
  503. },
  504. },
  505. result: []map[string]interface{}{{
  506. "a": 0.1,
  507. }},
  508. }, {
  509. sql: `SELECT json_path_query(equipment, "$.rings[? @.weight>1]") AS a FROM test`,
  510. data: &xsql.Tuple{
  511. Emitter: "test",
  512. Message: xsql.Message{
  513. "class": "warrior",
  514. "equipment": map[string]interface{}{
  515. "rings": []interface{}{
  516. map[string]interface{}{
  517. "name": "ring of despair",
  518. "weight": 0.1,
  519. }, map[string]interface{}{
  520. "name": "ring of strength",
  521. "weight": 2.4,
  522. },
  523. },
  524. "arm_right": "Sword of flame",
  525. "arm_left": "Shield of faith",
  526. },
  527. },
  528. },
  529. result: []map[string]interface{}{{
  530. "a": []interface{}{
  531. map[string]interface{}{
  532. "name": "ring of strength",
  533. "weight": 2.4,
  534. },
  535. },
  536. }},
  537. }, {
  538. sql: `SELECT json_path_query(equipment, "$.rings[? @.weight>1].name") AS a FROM test`,
  539. data: &xsql.Tuple{
  540. Emitter: "test",
  541. Message: xsql.Message{
  542. "class": "warrior",
  543. "equipment": map[string]interface{}{
  544. "rings": []interface{}{
  545. map[string]interface{}{
  546. "name": "ring of despair",
  547. "weight": 0.1,
  548. }, map[string]interface{}{
  549. "name": "ring of strength",
  550. "weight": 2.4,
  551. },
  552. },
  553. "arm_right": "Sword of flame",
  554. "arm_left": "Shield of faith",
  555. },
  556. },
  557. },
  558. result: []map[string]interface{}{{
  559. "a": []interface{}{
  560. "ring of strength",
  561. },
  562. }},
  563. }, {
  564. sql: `SELECT json_path_exists(equipment, "$.rings[? @.weight>5]") AS a FROM test`,
  565. data: &xsql.Tuple{
  566. Emitter: "test",
  567. Message: xsql.Message{
  568. "class": "warrior",
  569. "equipment": map[string]interface{}{
  570. "rings": []interface{}{
  571. map[string]interface{}{
  572. "name": "ring of despair",
  573. "weight": 0.1,
  574. }, map[string]interface{}{
  575. "name": "ring of strength",
  576. "weight": 2.4,
  577. },
  578. },
  579. "arm_right": "Sword of flame",
  580. "arm_left": "Shield of faith",
  581. },
  582. },
  583. },
  584. result: []map[string]interface{}{{
  585. "a": false,
  586. }},
  587. }, {
  588. sql: `SELECT json_path_exists(equipment, "$.ring1") AS a FROM test`,
  589. data: &xsql.Tuple{
  590. Emitter: "test",
  591. Message: xsql.Message{
  592. "class": "warrior",
  593. "equipment": map[string]interface{}{
  594. "rings": []interface{}{
  595. map[string]interface{}{
  596. "name": "ring of despair",
  597. "weight": 0.1,
  598. }, map[string]interface{}{
  599. "name": "ring of strength",
  600. "weight": 2.4,
  601. },
  602. },
  603. "arm_right": "Sword of flame",
  604. "arm_left": "Shield of faith",
  605. },
  606. },
  607. },
  608. result: []map[string]interface{}{{
  609. "a": false,
  610. }},
  611. }, {
  612. sql: `SELECT json_path_exists(equipment, "$.rings") AS a FROM test`,
  613. data: &xsql.Tuple{
  614. Emitter: "test",
  615. Message: xsql.Message{
  616. "class": "warrior",
  617. "equipment": map[string]interface{}{
  618. "rings": []interface{}{
  619. map[string]interface{}{
  620. "name": "ring of despair",
  621. "weight": 0.1,
  622. }, map[string]interface{}{
  623. "name": "ring of strength",
  624. "weight": 2.4,
  625. },
  626. },
  627. "arm_right": "Sword of flame",
  628. "arm_left": "Shield of faith",
  629. },
  630. },
  631. },
  632. result: []map[string]interface{}{{
  633. "a": true,
  634. }},
  635. }, {
  636. sql: `SELECT json_path_query(equipment, "$.rings[? (@.weight>1)].name") AS a FROM test`,
  637. data: &xsql.Tuple{
  638. Emitter: "test",
  639. Message: xsql.Message{
  640. "class": "warrior",
  641. "equipment": map[string]interface{}{
  642. "rings": []map[string]interface{}{
  643. {
  644. "name": "ring of despair",
  645. "weight": 0.1,
  646. }, {
  647. "name": "ring of strength",
  648. "weight": 2.4,
  649. },
  650. },
  651. "arm_right": "Sword of flame",
  652. "arm_left": "Shield of faith",
  653. },
  654. },
  655. },
  656. result: []map[string]interface{}{{
  657. "a": []interface{}{
  658. "ring of strength",
  659. },
  660. }},
  661. }, {
  662. sql: `SELECT json_path_query(equipment, "$.rings[*]") AS a FROM test`,
  663. data: &xsql.Tuple{
  664. Emitter: "test",
  665. Message: xsql.Message{
  666. "class": "warrior",
  667. "equipment": map[string]interface{}{
  668. "rings": []float64{
  669. 0.1, 2.4,
  670. },
  671. "arm_right": "Sword of flame",
  672. "arm_left": "Shield of faith",
  673. },
  674. },
  675. },
  676. result: []map[string]interface{}{{
  677. "a": []interface{}{
  678. 0.1, 2.4,
  679. },
  680. }},
  681. }, {
  682. sql: `SELECT json_path_query(equipment, "$.rings") AS a FROM test`,
  683. data: &xsql.Tuple{
  684. Emitter: "test",
  685. Message: xsql.Message{
  686. "class": "warrior",
  687. "equipment": map[string]interface{}{
  688. "rings": []float64{
  689. 0.1, 2.4,
  690. },
  691. "arm_right": "Sword of flame",
  692. "arm_left": "Shield of faith",
  693. },
  694. },
  695. },
  696. result: []map[string]interface{}{{
  697. "a": []interface{}{
  698. 0.1, 2.4,
  699. },
  700. }},
  701. }, {
  702. sql: `SELECT json_path_query(equipment, "$[0].rings[1]") AS a FROM test`,
  703. data: &xsql.Tuple{
  704. Emitter: "test",
  705. Message: xsql.Message{
  706. "class": "warrior",
  707. "equipment": []map[string]interface{}{
  708. {
  709. "rings": []float64{
  710. 0.1, 2.4,
  711. },
  712. "arm_right": "Sword of flame",
  713. "arm_left": "Shield of faith",
  714. },
  715. },
  716. },
  717. },
  718. result: []map[string]interface{}{{
  719. "a": 2.4,
  720. }},
  721. }, {
  722. sql: "SELECT json_path_query(equipment, \"$[0][\\\"arm.left\\\"]\") AS a FROM test",
  723. data: &xsql.Tuple{
  724. Emitter: "test",
  725. Message: xsql.Message{
  726. "class": "warrior",
  727. "equipment": []map[string]interface{}{
  728. {
  729. "rings": []float64{
  730. 0.1, 2.4,
  731. },
  732. "arm.right": "Sword of flame",
  733. "arm.left": "Shield of faith",
  734. },
  735. },
  736. },
  737. },
  738. result: []map[string]interface{}{{
  739. "a": "Shield of faith",
  740. }},
  741. }, {
  742. sql: "SELECT json_path_query(equipment, \"$[\\\"arm.left\\\"]\") AS a FROM test",
  743. data: &xsql.Tuple{
  744. Emitter: "test",
  745. Message: xsql.Message{
  746. "class": "warrior",
  747. "equipment": `{"rings": [0.1, 2.4],"arm.right": "Sword of flame","arm.left": "Shield of faith"}`,
  748. },
  749. },
  750. result: []map[string]interface{}{{
  751. "a": "Shield of faith",
  752. }},
  753. }, {
  754. sql: "SELECT json_path_query(equipment, \"$[0][\\\"arm.left\\\"]\") AS a FROM test",
  755. data: &xsql.Tuple{
  756. Emitter: "test",
  757. Message: xsql.Message{
  758. "class": "warrior",
  759. "equipment": `[{"rings": [0.1, 2.4],"arm.right": "Sword of flame","arm.left": "Shield of faith"}]`,
  760. },
  761. },
  762. result: []map[string]interface{}{{
  763. "a": "Shield of faith",
  764. }},
  765. }, {
  766. sql: `SELECT all[poi[-1] + 1]->ts as powerOnTs FROM test`,
  767. data: &xsql.Tuple{
  768. Emitter: "test",
  769. Message: xsql.Message{
  770. "all": []map[string]interface{}{
  771. {"SystemPowerMode": 0, "VehicleSpeed": 0, "FLWdwPosition": 0, "FrontWiperSwitchStatus": float64(1), "ts": 0},
  772. {"SystemPowerMode": 0, "VehicleSpeed": 0, "FLWdwPosition": 0, "FrontWiperSwitchStatus": float64(4), "ts": 500},
  773. {"SystemPowerMode": 2, "VehicleSpeed": 0, "FLWdwPosition": 0, "FrontWiperSwitchStatus": 0, "ts": 1000},
  774. {"SystemPowerMode": 2, "VehicleSpeed": 10, "FLWdwPosition": 20, "FrontWiperSwitchStatus": 0, "ts": 60000},
  775. {"SystemPowerMode": 2, "VehicleSpeed": 10, "FLWdwPosition": 20, "FrontWiperSwitchStatus": 0, "ts": 89500},
  776. {"SystemPowerMode": 2, "VehicleSpeed": 20, "FLWdwPosition": 50, "FrontWiperSwitchStatus": 5, "ts": 90000},
  777. {"SystemPowerMode": 2, "VehicleSpeed": 40, "FLWdwPosition": 60, "FrontWiperSwitchStatus": 5, "ts": 121000},
  778. },
  779. "poi": []interface{}{0, 1},
  780. },
  781. },
  782. result: []map[string]interface{}{{
  783. "powerOnTs": float64(1000),
  784. }},
  785. }, {
  786. sql: `SELECT json_path_query(equipment, "$.arm_right") AS a FROM test`,
  787. data: &xsql.Tuple{
  788. Emitter: "test",
  789. Message: xsql.Message{
  790. "class": "warrior",
  791. "equipment2": map[string]interface{}{
  792. "rings": []map[string]interface{}{
  793. {
  794. "name": "ring of despair",
  795. "weight": 0.1,
  796. }, {
  797. "name": "ring of strength",
  798. "weight": 2.4,
  799. },
  800. },
  801. "arm_right": "Sword of flame",
  802. "arm_left": "Shield of faith",
  803. },
  804. },
  805. },
  806. err: "run Select error: call func json_path_query error: json_path_query function error: the first argument must be a map but got nil",
  807. },
  808. }
  809. fmt.Printf("The test bucket size is %d.\n\n", len(tests))
  810. contextLogger := conf.Log.WithField("rule", "TestJsonFunc_Apply1")
  811. ctx := context.WithValue(context.Background(), context.LoggerKey, contextLogger)
  812. for i, tt := range tests {
  813. stmt, err := xsql.NewParser(strings.NewReader(tt.sql)).Parse()
  814. if err != nil || stmt == nil {
  815. t.Errorf("parse sql %s error %v", tt.sql, err)
  816. }
  817. pp := &ProjectOp{Fields: stmt.Fields}
  818. fv, afv := xsql.NewFunctionValuersForOp(nil, xsql.FuncRegisters)
  819. result := pp.Apply(ctx, tt.data, fv, afv)
  820. switch rt := result.(type) {
  821. case []byte:
  822. if tt.err == "" {
  823. var mapRes []map[string]interface{}
  824. err := json.Unmarshal(rt, &mapRes)
  825. if err != nil {
  826. t.Errorf("Failed to parse the input into map.\n")
  827. continue
  828. }
  829. if !reflect.DeepEqual(tt.result, mapRes) {
  830. t.Errorf("%d. %q\n\nresult mismatch:\n\nexp=%#v\n\ngot=%#v\n\n", i, tt.sql, tt.result, mapRes)
  831. }
  832. } else {
  833. t.Errorf("%d: invalid result:\n exp error %s\n got=%s\n\n", i, tt.err, result)
  834. }
  835. case error:
  836. if tt.err == "" {
  837. t.Errorf("%d: got error:\n exp=%s\n got=%s\n\n", i, tt.result, err)
  838. } else if !reflect.DeepEqual(tt.err, testx.Errstring(rt)) {
  839. t.Errorf("%d: error mismatch:\n exp=%s\n got=%s\n\n", i, tt.err, err)
  840. }
  841. default:
  842. t.Errorf("%d: Invalid returned result found %v", i, result)
  843. }
  844. }
  845. }