migration_test.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081
  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 util
  15. import (
  16. "github.com/lf-edge/ekuiper/pkg/kv"
  17. "github.com/patrickmn/go-cache"
  18. "os"
  19. "path"
  20. "path/filepath"
  21. "testing"
  22. )
  23. func TestDataMigration(t *testing.T) {
  24. kvs := make(map[string]string)
  25. kvs["mqtt"] = `{"sql":"create stream mqtt(age BIGINT) WITH (DATASOURCE = \"dev/+/msg\", FORMAT = \"json\");"}`
  26. kvs["log"] = `{"id":"log","sql":"SELECT * FROM mqtt","actions":[{"log":{}}]}`
  27. dir, _ := filepath.Abs("testMigration")
  28. dirSqlite, _ := filepath.Split(dir)
  29. fpath := path.Join(dir, "stores.data")
  30. if f, _ := os.Stat(fpath); f != nil {
  31. os.Remove(fpath)
  32. }
  33. if _, err := os.Stat(dir); os.IsNotExist(err) {
  34. os.MkdirAll(dir, os.ModePerm)
  35. }
  36. defer os.RemoveAll(dir)
  37. defer os.Remove(path.Join(dirSqlite, "sqliteKV.db"))
  38. c := cache.New(cache.NoExpiration, 0)
  39. if err := gocacheOpen(c, fpath); nil != err {
  40. t.Error(err)
  41. return
  42. }
  43. for k, v := range kvs {
  44. if err := gocacheSet(c, fpath, k, v); nil != err {
  45. t.Error(err)
  46. return
  47. }
  48. }
  49. if err := gocacheClose(c, fpath); nil != err {
  50. t.Error(err)
  51. }
  52. if err := DataMigration(dir); nil != err {
  53. t.Error(err)
  54. return
  55. }
  56. store := kv.GetDefaultKVStore(dir)
  57. if err := store.Open(); nil != err {
  58. t.Error(err)
  59. return
  60. }
  61. defer store.Close()
  62. for k, v := range kvs {
  63. var dbVal string
  64. if ok, _ := store.Get(k, &dbVal); !ok {
  65. t.Error("not found key ", k)
  66. return
  67. } else if v != dbVal {
  68. t.Error("gocache save:", v, "sqlite save:", dbVal)
  69. return
  70. }
  71. }
  72. }