reformat
diff --git a/data.go b/data.go
index eca1491..3d33902 100644
--- a/data.go
+++ b/data.go
@@ -71,6 +71,17 @@
type dbManager struct {
db apid.DB
dbMux sync.RWMutex
+ data apid.DataService
+}
+
+func (dbc *dbManager) setDbVersion(version string) {
+ db, err := dbc.data.DBVersion(version)
+ if err != nil {
+ log.Panicf("Unable to access database: %v", err)
+ }
+ dbc.dbMux.Lock()
+ dbc.db = db
+ dbc.dbMux.Unlock()
}
func (dbc *dbManager) getDb() apid.DB {
diff --git a/data_test.go b/data_test.go
index e1bf8dc..f6115d4 100644
--- a/data_test.go
+++ b/data_test.go
@@ -15,55 +15,67 @@
package apidApigeeSync
import (
+ "github.com/30x/apid-core/data"
"github.com/apigee-labs/transicator/common"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
+ "io/ioutil"
+ "reflect"
"sort"
+ "strconv"
"sync"
)
-var _ = Describe("data access tests", func() {
- var testCount int
+var _ = Describe("DB manager tests", func() {
var testDbMan *dbManager
+ var testCount int
BeforeEach(func() {
- testCount += 1
testDbMan = &dbManager{
dbMux: sync.RWMutex{},
+ data: dataService,
}
+ testCount += 1
- db := testDbMan.getDb()
-
- //all tests in this file operate on the api_product table. Create the necessary tables for this here
- db.Exec("CREATE TABLE _transicator_tables " +
- "(tableName varchar not null, columnName varchar not null, " +
- "typid integer, primaryKey bool);")
- db.Exec("DELETE from _transicator_tables")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','id',2950,1)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','tenant_id',1043,1)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','description',1043,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','api_resources',1015,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','approval_type',1043,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','scopes',1015,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','proxies',1015,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','environments',1015,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_at',1114,1)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_by',1043,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_at',1114,1)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_by',1043,0)")
- db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','_change_selector',1043,0)")
-
- db.Exec("CREATE TABLE kms_api_product (id text,tenant_id text,name text, description text, " +
- "api_resources text,approval_type text,scopes text,proxies text, environments text," +
- "created_at blob, created_by text,updated_at blob,updated_by text,_change_selector text, " +
- "primary key (id,tenant_id,created_at,updated_at));")
- db.Exec("DELETE from kms_api_product")
-
- setDB(db)
- initDB(db)
-
+ testDbMan.setDbVersion("data_test_" + strconv.Itoa(testCount))
})
- Context("Update processing", func() {
+ var _ = AfterEach(func() {
+ testDbMan = nil
+ data.Delete(data.VersionedDBID("common", "data_test_"+strconv.Itoa(testCount)))
+ })
+
+ Context("Basic Update/Insert/Delete processing", func() {
+ BeforeEach(func() {
+ db := testDbMan.getDb()
+
+ //all tests in this file operate on the api_product table. Create the necessary tables for this here
+ db.Exec("CREATE TABLE _transicator_tables " +
+ "(tableName varchar not null, columnName varchar not null, " +
+ "typid integer, primaryKey bool);")
+ db.Exec("DELETE from _transicator_tables")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','id',2950,1)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','tenant_id',1043,1)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','description',1043,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','api_resources',1015,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','approval_type',1043,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','scopes',1015,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','proxies',1015,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','environments',1015,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_at',1114,1)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','created_by',1043,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_at',1114,1)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','updated_by',1043,0)")
+ db.Exec("INSERT INTO _transicator_tables VALUES('kms_api_product','_change_selector',1043,0)")
+
+ db.Exec("CREATE TABLE kms_api_product (id text,tenant_id text,name text, description text, " +
+ "api_resources text,approval_type text,scopes text,proxies text, environments text," +
+ "created_at blob, created_by text,updated_at blob,updated_by text,_change_selector text, " +
+ "primary key (id,tenant_id,created_at,updated_at));")
+ db.Exec("DELETE from kms_api_product")
+
+ testDbMan.initDefaultDb()
+ })
+
It("unit test buildUpdateSql with single primary key", func() {
testRow := common.Row{
"id": {
@@ -193,9 +205,9 @@
},
}
//insert and assert success
- Expect(true).To(Equal(processChangeList(event)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
var nRows int
- err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+ err := testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
@@ -210,13 +222,13 @@
}
//do the update
- Expect(true).To(Equal(processChangeList(event)))
- err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
+ err = testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
@@ -288,9 +300,9 @@
},
}
//insert and assert success
- Expect(true).To(Equal(processChangeList(event)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
var nRows int
- err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+ err := testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
@@ -305,13 +317,13 @@
}
//do the update
- Expect(true).To(Equal(processChangeList(event)))
- err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='new_id' and description='new description'").Scan(&nRows)
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
+ err = testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='new_id' and description='new description'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
})
@@ -390,9 +402,9 @@
},
}
//insert and assert success
- Expect(true).To(Equal(processChangeList(event)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
var nRows int
- err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+ err := testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
@@ -407,13 +419,13 @@
}
//do the update
- Expect(true).To(Equal(processChangeList(event)))
- err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
+ err = testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
})
@@ -489,9 +501,9 @@
},
}
//insert and assert success
- Expect(true).To(Equal(processChangeList(event)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
var nRows int
- err := getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
+ err := testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='A product for testing Greg'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
@@ -506,19 +518,17 @@
}
//do the update
- Expect(true).To(Equal(processChangeList(event)))
- err = getDB().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
+ err = testDbMan.getDb().QueryRow("SELECT count(id) FROM kms_api_product WHERE id='87a4bfaa-b3c4-47cd-b6c5-378cdb68610c' and description='new description'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
})
- })
- Context("Insert processing", func() {
It("Properly constructs insert sql for one row", func() {
newRow := common.Row{
"id": {
@@ -659,16 +669,16 @@
},
}
- Expect(true).To(Equal(processChangeList(event)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
var nRows int
- err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+ err := testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
@@ -743,22 +753,22 @@
},
}
- Expect(true).To(Equal(processChangeList(event)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event)))
var nRows int
- err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+ err := testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(2))
})
@@ -801,12 +811,12 @@
},
}
- ok := processChangeList(event)
+ ok := testDbMan.writeTransaction(event)
Expect(false).To(Equal(ok))
var nRows int
//assert that no extraneous rows were added
- err := getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err := testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(0))
})
@@ -880,12 +890,10 @@
},
}
- ok := processChangeList(event)
+ ok := testDbMan.writeTransaction(event)
Expect(false).To(Equal(ok))
})
- })
- Context("Delete processing", func() {
It("Properly constructs sql prepare for Delete", func() {
row := common.Row{
"id": {
@@ -914,7 +922,7 @@
},
}
- pkeys, err := getPkeysForTable("kms_api_product")
+ pkeys, err := testDbMan.getPkeysForTable("kms_api_product")
Expect(err).Should(Succeed())
sql := buildDeleteSql("kms_api_product", row, pkeys)
Expect(sql).To(Equal("DELETE FROM kms_api_product WHERE created_at=$1 AND id=$2 AND tenant_id=$3 AND updated_at=$4"))
@@ -966,28 +974,28 @@
},
}
- Expect(true).To(Equal(processChangeList(event1)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event1)))
var nRows int
- err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+ err := testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
- Expect(true).To(Equal(processChangeList(event2)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event2)))
// validate delete
- err = getDB().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(0))
// delete again should fail - coz entry will not exist
- Expect(false).To(Equal(processChangeList(event2)))
+ Expect(false).To(Equal(testDbMan.writeTransaction(event2)))
})
It("verify multiple insert and single delete works", func() {
@@ -1068,43 +1076,43 @@
},
}
- Expect(true).To(Equal(processChangeList(event1)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event1)))
var nRows int
//verify first row
- err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+ err := testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//verify second row
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(2))
- Expect(true).To(Equal(processChangeList(event2)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event2)))
//verify second row still exists
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='b' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
// validate delete
- err = getDB().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("select count(*) from kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
// delete again should fail - coz entry will not exist
- Expect(false).To(Equal(processChangeList(event2)))
+ Expect(false).To(Equal(testDbMan.writeTransaction(event2)))
}, 3)
It("verify single insert and multiple delete fails", func() {
@@ -1185,21 +1193,209 @@
},
}
- Expect(true).To(Equal(processChangeList(event1)))
+ Expect(true).To(Equal(testDbMan.writeTransaction(event1)))
var nRows int
//verify insert
- err := getDB().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
+ err := testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product WHERE id='a' and api_resources='r'" +
"and environments='{test}' and tenant_id='t' and description='d' and created_at='c' and updated_at='u'" +
"and _change_selector='cs'").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
//assert that no extraneous rows were added
- err = getDB().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
+ err = testDbMan.getDb().QueryRow("SELECT count(*) FROM kms_api_product").Scan(&nRows)
Expect(err).NotTo(HaveOccurred())
Expect(nRows).To(Equal(1))
- Expect(false).To(Equal(processChangeList(event2)))
+ Expect(false).To(Equal(testDbMan.writeTransaction(event2)))
+
+ }, 3)
+ })
+
+ Context("ApigeeSync methods", func() {
+ var createTestDb = func(sqlfile string) {
+
+ db := testDbMan.getDb()
+ sqlStatementsBuffer, err := ioutil.ReadFile(sqlfile)
+ Expect(err).Should(Succeed())
+ sqlStatementsString := string(sqlStatementsBuffer)
+ _, err = db.Exec(sqlStatementsString)
+ Expect(err).Should(Succeed())
+ }
+
+ It("test getClusterCount", func() {
+ createTestDb("./sql/init_listener_test_duplicate_apids.sql")
+ Expect(testDbMan.getClusterCount()).To(Equal(2))
+ })
+
+ It("should process a valid Snapshot", func() {
+
+ createTestDb("./sql/init_listener_test_valid_snapshot.sql")
+
+ dbVersion := "data_test_" + strconv.Itoa(testCount)
+ err := testDbMan.updateApidInstanceInfo()
+ Expect(err).Should(Succeed())
+
+ info, err := testDbMan.getApidInstanceInfo()
+ Expect(err).Should(Succeed())
+ Expect(info.LastSnapshot).To(Equal(dbVersion))
+
+ db := testDbMan.getDb()
+
+ // apid Cluster
+ var dcs []dataApidCluster
+ rows, err := db.Query(`
+ SELECT id, name, description, umbrella_org_app_name,
+ created, created_by, updated, updated_by
+ FROM EDGEX_APID_CLUSTER`)
+ Expect(err).NotTo(HaveOccurred())
+ defer rows.Close()
+
+ c := dataApidCluster{}
+ for rows.Next() {
+ rows.Scan(&c.ID, &c.Name, &c.Description, &c.OrgAppName,
+ &c.Created, &c.CreatedBy, &c.Updated, &c.UpdatedBy)
+ dcs = append(dcs, c)
+ }
+
+ Expect(len(dcs)).To(Equal(1))
+ dc := dcs[0]
+
+ Expect(dc.ID).To(Equal("i"))
+ Expect(dc.Name).To(Equal("n"))
+ Expect(dc.Description).To(Equal("d"))
+ Expect(dc.OrgAppName).To(Equal("o"))
+ Expect(dc.Created).To(Equal("c"))
+ Expect(dc.CreatedBy).To(Equal("c"))
+ Expect(dc.Updated).To(Equal("u"))
+ Expect(dc.UpdatedBy).To(Equal("u"))
+
+ // Data Scope
+ var dds []dataDataScope
+
+ rows, err = db.Query(`
+ SELECT id, apid_cluster_id, scope, org,
+ env, created, created_by, updated,
+ updated_by
+ FROM EDGEX_DATA_SCOPE`)
+ Expect(err).NotTo(HaveOccurred())
+ defer rows.Close()
+
+ d := dataDataScope{}
+ for rows.Next() {
+ rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
+ &d.Env, &d.Created, &d.CreatedBy, &d.Updated,
+ &d.UpdatedBy)
+ dds = append(dds, d)
+ }
+
+ Expect(len(dds)).To(Equal(3))
+ ds := dds[0]
+
+ Expect(ds.ID).To(Equal("i"))
+ Expect(ds.Org).To(Equal("o"))
+ Expect(ds.Env).To(Equal("e1"))
+ Expect(ds.Scope).To(Equal("s1"))
+ Expect(ds.Created).To(Equal("c"))
+ Expect(ds.CreatedBy).To(Equal("c"))
+ Expect(ds.Updated).To(Equal("u"))
+ Expect(ds.UpdatedBy).To(Equal("u"))
+
+ ds = dds[1]
+ Expect(ds.Env).To(Equal("e2"))
+ Expect(ds.Scope).To(Equal("s1"))
+ ds = dds[2]
+ Expect(ds.Env).To(Equal("e3"))
+ Expect(ds.Scope).To(Equal("s2"))
+
+ scopes := testDbMan.findScopesForId("a")
+ Expect(len(scopes)).To(Equal(6))
+ expectedScopes := []string{"s1", "s2", "org_scope_1", "env_scope_1", "env_scope_2", "env_scope_3"}
+ sort.Strings(scopes)
+ sort.Strings(expectedScopes)
+ Expect(reflect.DeepEqual(scopes, expectedScopes)).To(BeTrue())
+ }, 3)
+
+ It("insert event should add", func() {
+ createTestDb("./sql/init_listener_test_no_datascopes.sql", "test_changes_insert")
+ event := common.ChangeList{
+ LastSequence: "test",
+ Changes: []common.Change{
+ {
+ Operation: common.Insert,
+ Table: LISTENER_TABLE_DATA_SCOPE,
+ NewRow: common.Row{
+ "id": &common.ColumnVal{Value: "i"},
+ "apid_cluster_id": &common.ColumnVal{Value: "a"},
+ "scope": &common.ColumnVal{Value: "s1"},
+ "org": &common.ColumnVal{Value: "o"},
+ "env": &common.ColumnVal{Value: "e"},
+ "created": &common.ColumnVal{Value: "c"},
+ "created_by": &common.ColumnVal{Value: "c"},
+ "updated": &common.ColumnVal{Value: "u"},
+ "updated_by": &common.ColumnVal{Value: "u"},
+ "_change_selector": &common.ColumnVal{Value: "cs"},
+ },
+ },
+ {
+ Operation: common.Insert,
+ Table: LISTENER_TABLE_DATA_SCOPE,
+ NewRow: common.Row{
+ "id": &common.ColumnVal{Value: "j"},
+ "apid_cluster_id": &common.ColumnVal{Value: "a"},
+ "scope": &common.ColumnVal{Value: "s2"},
+ "org": &common.ColumnVal{Value: "o"},
+ "env": &common.ColumnVal{Value: "e"},
+ "created": &common.ColumnVal{Value: "c"},
+ "created_by": &common.ColumnVal{Value: "c"},
+ "updated": &common.ColumnVal{Value: "u"},
+ "updated_by": &common.ColumnVal{Value: "u"},
+ "_change_selector": &common.ColumnVal{Value: "cs"},
+ },
+ },
+ },
+ }
+
+ testDbMan.writeTransaction(&event)
+
+ var dds []dataDataScope
+
+ rows, err := testDbMan.getDb().Query(`
+ SELECT id, apid_cluster_id, scope, org,
+ env, created, created_by, updated,
+ updated_by
+ FROM EDGEX_DATA_SCOPE`)
+ Expect(err).NotTo(HaveOccurred())
+ defer rows.Close()
+
+ d := dataDataScope{}
+ for rows.Next() {
+ rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
+ &d.Env, &d.Created, &d.CreatedBy, &d.Updated,
+ &d.UpdatedBy)
+ dds = append(dds, d)
+ }
+
+ //three already existing
+ Expect(len(dds)).To(Equal(2))
+ ds := dds[0]
+
+ Expect(ds.ID).To(Equal("i"))
+ Expect(ds.Org).To(Equal("o"))
+ Expect(ds.Env).To(Equal("e"))
+ Expect(ds.Scope).To(Equal("s1"))
+ Expect(ds.Created).To(Equal("c"))
+ Expect(ds.CreatedBy).To(Equal("c"))
+ Expect(ds.Updated).To(Equal("u"))
+ Expect(ds.UpdatedBy).To(Equal("u"))
+
+ ds = dds[1]
+ Expect(ds.Scope).To(Equal("s2"))
+
+ scopes := testDbMan.findScopesForId("a")
+ Expect(len(scopes)).To(Equal(2))
+ Expect(scopes[0]).To(Equal("s1"))
+ Expect(scopes[1]).To(Equal("s2"))
}, 3)
})
diff --git a/init.go b/init.go
index d8226de..f456953 100644
--- a/init.go
+++ b/init.go
@@ -105,6 +105,7 @@
dbMan = &dbManager{
dbMux: sync.RWMutex{},
+ data: services.Data(),
}
// set up default database
diff --git a/listener.go b/listener.go
index 76ec295..cffeb05 100644
--- a/listener.go
+++ b/listener.go
@@ -15,7 +15,6 @@
package apidApigeeSync
import (
- "github.com/30x/apid-core"
"github.com/apigee-labs/transicator/common"
)
@@ -31,26 +30,21 @@
func (lm *listenerManager) processSnapshot(snapshot *common.Snapshot) {
log.Debugf("Snapshot received. Switching to DB version: %s", snapshot.SnapshotInfo)
- db, err := dataService.DBVersion(snapshot.SnapshotInfo)
- if err != nil {
- log.Panicf("Unable to access database: %v", err)
- }
-
- lm.processSqliteSnapshot(db)
+ lm.dbm.setDbVersion(snapshot.SnapshotInfo)
+ lm.processSqliteSnapshot()
//update apid instance info
apidInfo.LastSnapshot = snapshot.SnapshotInfo
- err = lm.dbm.updateApidInstanceInfo()
+ err := lm.dbm.updateApidInstanceInfo()
if err != nil {
log.Panicf("Unable to update instance info: %v", err)
}
- lm.dbm.setDb(db)
log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo)
}
-func (lm *listenerManager) processSqliteSnapshot(db apid.DB) {
+func (lm *listenerManager) processSqliteSnapshot() {
if count, err := lm.dbm.getClusterCount(); err != nil || count != 1 {
log.Panicf("Illegal state for apid_cluster. Must be a single row. %v", err)
diff --git a/listener_test.go b/listener_test.go
index d48c70e..2950288 100644
--- a/listener_test.go
+++ b/listener_test.go
@@ -20,11 +20,19 @@
"github.com/apigee-labs/transicator/common"
"os"
- "reflect"
- "sort"
)
var _ = Describe("listener", func() {
+ var testListenerMan *listenerManager
+ var mockDbMan *dummyDbMan
+ var testCount int
+ BeforeEach(func() {
+ mockDbMan = &dummyDbMan{}
+ testListenerMan = &listenerManager{
+ dbm: mockDbMan,
+ }
+ testCount += 1
+ })
var createTestDb = func(sqlfile string, dbId string) common.Snapshot {
initDb(sqlfile, "./mockdb.sqlite3")
@@ -40,11 +48,11 @@
Context("ApigeeSync snapshot event", func() {
It("should fail if more than one apid_cluster rows", func() {
- event := createTestDb("./sql/init_listener_test_duplicate_apids.sql", "test_snapshot_fail_multiple_clusters")
- Expect(func() { processSnapshot(&event) }).To(Panic())
+ mockDbMan.clusterCount = 2
+ Expect(func() { testListenerMan.processSnapshot(&common.Snapshot{}) }).To(Panic())
}, 3)
- It("should fail if more than one apid_cluster rows", func() {
+ It("test scope change", func() {
newScopes := []string{"foo"}
scopes := []string{"bar"}
Expect(scopeChanged(newScopes, scopes)).To(Equal(changeServerError{Code: "Scope changes detected; must get new snapshot"}))
@@ -59,99 +67,6 @@
Expect(scopeChanged(newScopes, scopes)).To(BeNil())
}, 3)
-
- It("should process a valid Snapshot", func() {
-
- event := createTestDb("./sql/init_listener_test_valid_snapshot.sql", "test_snapshot_valid")
-
- processSnapshot(&event)
-
- info, err := getApidInstanceInfo()
- Expect(err).NotTo(HaveOccurred())
-
- Expect(info.LastSnapshot).To(Equal(event.SnapshotInfo))
-
- db := getDB()
-
- expectedDB, err := dataService.DBVersion(event.SnapshotInfo)
- Expect(err).NotTo(HaveOccurred())
-
- Expect(db == expectedDB).Should(BeTrue())
-
- // apid Cluster
- var dcs []dataApidCluster
-
- rows, err := db.Query(`
- SELECT id, name, description, umbrella_org_app_name,
- created, created_by, updated, updated_by
- FROM EDGEX_APID_CLUSTER`)
- Expect(err).NotTo(HaveOccurred())
- defer rows.Close()
-
- c := dataApidCluster{}
- for rows.Next() {
- rows.Scan(&c.ID, &c.Name, &c.Description, &c.OrgAppName,
- &c.Created, &c.CreatedBy, &c.Updated, &c.UpdatedBy)
- dcs = append(dcs, c)
- }
-
- Expect(len(dcs)).To(Equal(1))
- dc := dcs[0]
-
- Expect(dc.ID).To(Equal("i"))
- Expect(dc.Name).To(Equal("n"))
- Expect(dc.Description).To(Equal("d"))
- Expect(dc.OrgAppName).To(Equal("o"))
- Expect(dc.Created).To(Equal("c"))
- Expect(dc.CreatedBy).To(Equal("c"))
- Expect(dc.Updated).To(Equal("u"))
- Expect(dc.UpdatedBy).To(Equal("u"))
-
- // Data Scope
- var dds []dataDataScope
-
- rows, err = db.Query(`
- SELECT id, apid_cluster_id, scope, org,
- env, created, created_by, updated,
- updated_by
- FROM EDGEX_DATA_SCOPE`)
- Expect(err).NotTo(HaveOccurred())
- defer rows.Close()
-
- d := dataDataScope{}
- for rows.Next() {
- rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
- &d.Env, &d.Created, &d.CreatedBy, &d.Updated,
- &d.UpdatedBy)
- dds = append(dds, d)
- }
-
- Expect(len(dds)).To(Equal(3))
- ds := dds[0]
-
- Expect(ds.ID).To(Equal("i"))
- Expect(ds.Org).To(Equal("o"))
- Expect(ds.Env).To(Equal("e1"))
- Expect(ds.Scope).To(Equal("s1"))
- Expect(ds.Created).To(Equal("c"))
- Expect(ds.CreatedBy).To(Equal("c"))
- Expect(ds.Updated).To(Equal("u"))
- Expect(ds.UpdatedBy).To(Equal("u"))
-
- ds = dds[1]
- Expect(ds.Env).To(Equal("e2"))
- Expect(ds.Scope).To(Equal("s1"))
- ds = dds[2]
- Expect(ds.Env).To(Equal("e3"))
- Expect(ds.Scope).To(Equal("s2"))
-
- scopes := findScopesForId("a")
- Expect(len(scopes)).To(Equal(6))
- expectedScopes := []string{"s1", "s2", "org_scope_1", "env_scope_1", "env_scope_2", "env_scope_3"}
- sort.Strings(scopes)
- sort.Strings(expectedScopes)
- Expect(reflect.DeepEqual(scopes, expectedScopes)).To(BeTrue())
- }, 3)
})
Context("ApigeeSync change event", func() {
@@ -159,11 +74,6 @@
Context(LISTENER_TABLE_APID_CLUSTER, func() {
It("insert event should panic", func() {
- ssEvent := createTestDb("./sql/init_listener_test_valid_snapshot.sql", "test_changes_insert_panic")
- processSnapshot(&ssEvent)
-
- //save the last snapshot, so we can restore it at the end of this context
-
csEvent := common.ChangeList{
LastSequence: "test",
Changes: []common.Change{
@@ -174,13 +84,10 @@
},
}
- Expect(func() { processChangeList(&csEvent) }).To(Panic())
+ Expect(func() { testListenerMan.processChangeList(&csEvent) }).To(Panic())
}, 3)
It("update event should panic", func() {
- ssEvent := createTestDb("./sql/init_listener_test_valid_snapshot.sql", "test_changes_update_panic")
- processSnapshot(&ssEvent)
-
event := common.ChangeList{
LastSequence: "test",
Changes: []common.Change{
@@ -191,7 +98,7 @@
},
}
- Expect(func() { processChangeList(&event) }).To(Panic())
+ Expect(func() { testListenerMan.processChangeList(&event) }).To(Panic())
//restore the last snapshot
}, 3)
@@ -199,91 +106,6 @@
Context(LISTENER_TABLE_DATA_SCOPE, func() {
- It("insert event should add", func() {
- ssEvent := createTestDb("./sql/init_listener_test_no_datascopes.sql", "test_changes_insert")
- processSnapshot(&ssEvent)
-
- event := common.ChangeList{
- LastSequence: "test",
- Changes: []common.Change{
- {
- Operation: common.Insert,
- Table: LISTENER_TABLE_DATA_SCOPE,
- NewRow: common.Row{
- "id": &common.ColumnVal{Value: "i"},
- "apid_cluster_id": &common.ColumnVal{Value: "a"},
- "scope": &common.ColumnVal{Value: "s1"},
- "org": &common.ColumnVal{Value: "o"},
- "env": &common.ColumnVal{Value: "e"},
- "created": &common.ColumnVal{Value: "c"},
- "created_by": &common.ColumnVal{Value: "c"},
- "updated": &common.ColumnVal{Value: "u"},
- "updated_by": &common.ColumnVal{Value: "u"},
- "_change_selector": &common.ColumnVal{Value: "cs"},
- },
- },
- {
- Operation: common.Insert,
- Table: LISTENER_TABLE_DATA_SCOPE,
- NewRow: common.Row{
- "id": &common.ColumnVal{Value: "j"},
- "apid_cluster_id": &common.ColumnVal{Value: "a"},
- "scope": &common.ColumnVal{Value: "s2"},
- "org": &common.ColumnVal{Value: "o"},
- "env": &common.ColumnVal{Value: "e"},
- "created": &common.ColumnVal{Value: "c"},
- "created_by": &common.ColumnVal{Value: "c"},
- "updated": &common.ColumnVal{Value: "u"},
- "updated_by": &common.ColumnVal{Value: "u"},
- "_change_selector": &common.ColumnVal{Value: "cs"},
- },
- },
- },
- }
-
- processChangeList(&event)
-
- var dds []dataDataScope
-
- rows, err := getDB().Query(`
- SELECT id, apid_cluster_id, scope, org,
- env, created, created_by, updated,
- updated_by
- FROM EDGEX_DATA_SCOPE`)
- Expect(err).NotTo(HaveOccurred())
- defer rows.Close()
-
- d := dataDataScope{}
- for rows.Next() {
- rows.Scan(&d.ID, &d.ClusterID, &d.Scope, &d.Org,
- &d.Env, &d.Created, &d.CreatedBy, &d.Updated,
- &d.UpdatedBy)
- dds = append(dds, d)
- }
-
- //three already existing
- Expect(len(dds)).To(Equal(2))
- ds := dds[0]
-
- Expect(ds.ID).To(Equal("i"))
- Expect(ds.Org).To(Equal("o"))
- Expect(ds.Env).To(Equal("e"))
- Expect(ds.Scope).To(Equal("s1"))
- Expect(ds.Created).To(Equal("c"))
- Expect(ds.CreatedBy).To(Equal("c"))
- Expect(ds.Updated).To(Equal("u"))
- Expect(ds.UpdatedBy).To(Equal("u"))
-
- ds = dds[1]
- Expect(ds.Scope).To(Equal("s2"))
-
- scopes := findScopesForId("a")
- Expect(len(scopes)).To(Equal(2))
- Expect(scopes[0]).To(Equal("s1"))
- Expect(scopes[1]).To(Equal("s2"))
-
- }, 3)
-
It("delete event should delete", func() {
ssEvent := createTestDb("./sql/init_listener_test_no_datascopes.sql", "test_changes_delete")
processSnapshot(&ssEvent)
diff --git a/managerInterfaces.go b/managerInterfaces.go
index e406784..ff29c2d 100644
--- a/managerInterfaces.go
+++ b/managerInterfaces.go
@@ -61,10 +61,11 @@
getClusterCount() (numApidClusters int, err error)
alterClusterTable() (err error)
writeTransaction(*common.ChangeList) bool
+ setDbVersion(version string)
}
type listenerManagerInterface interface {
processSnapshot(snapshot *common.Snapshot)
- processSqliteSnapshot(db apid.DB)
+ processSqliteSnapshot()
processChangeList(changes *common.ChangeList) bool
}
diff --git a/token_test.go b/token_test.go
index e75b54d..cb38125 100644
--- a/token_test.go
+++ b/token_test.go
@@ -216,10 +216,13 @@
})
})
-type dummyDbMan struct{}
+type dummyDbMan struct {
+ clusterCount int
+ err error
+}
func (d *dummyDbMan) initDefaultDb() error {
- return nil
+ return d.err
}
func (d *dummyDbMan) setDb(apid.DB) {}
@@ -241,7 +244,7 @@
}
func (d *dummyDbMan) getPkeysForTable(tableName string) ([]string, error) {
- return nil, nil
+ return nil, d.err
}
func (d *dummyDbMan) findScopesForId(configId string) []string {
@@ -249,15 +252,15 @@
}
func (d *dummyDbMan) getDefaultDb() (apid.DB, error) {
- return nil, nil
+ return nil, d.err
}
func (d *dummyDbMan) updateApidInstanceInfo() error {
- return nil
+ return d.err
}
-func (d *dummyDbMan) getApidInstanceInfo() (info apidInstanceInfo, err error) {
- return
+func (d *dummyDbMan) getApidInstanceInfo() (apidInstanceInfo, error) {
+ return apidInstanceInfo{}, d.err
}
func (d *dummyDbMan) getLastSequence() string {
@@ -265,13 +268,20 @@
}
func (d *dummyDbMan) updateLastSequence(lastSequence string) error {
- return nil
+ return d.err
}
func (d *dummyDbMan) getClusterCount() (int, error) {
- return 1, nil
+ return d.clusterCount, d.err
}
func (d *dummyDbMan) alterClusterTable() error {
- return nil
+ return d.err
+}
+func (d *dummyDbMan) writeTransaction(_ *common.ChangeList) bool {
+ return true
+}
+
+func (d *dummyDbMan) setDbVersion(version string) {
+
}
