Support update data_scope, more testing
diff --git a/apigeeSync_suite_test.go b/apigeeSync_suite_test.go index 95a7da6..93c70fb 100644 --- a/apigeeSync_suite_test.go +++ b/apigeeSync_suite_test.go
@@ -10,12 +10,18 @@ "io/ioutil" "net/http/httptest" "os" + "encoding/json" + "net/http" + "github.com/apigee-labs/transicator/common" + "time" + "strconv" ) var ( tmpDir string testServer *httptest.Server testRouter apid.Router + phase int ) const testScope = "bootstrap" @@ -44,10 +50,241 @@ config.Set(configConsumerKey, "XXXXXXX") config.Set(configConsumerSecret, "YYYYYYY") - apid.InitializePlugins() + // mock upstream testServer + testRouter.HandleFunc("/accesstoken", func(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + + Expect(req.Method).To(Equal("POST")) + Expect(req.Header.Get("Content-Type")).To(Equal("application/x-www-form-urlencoded; param=value")) + + err := req.ParseForm() + Expect(err).NotTo(HaveOccurred()) + Expect(req.Form.Get("grant_type")).To(Equal("client_credentials")) + Expect(req.Header.Get("status")).To(Equal("ONLINE")) + Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) + Expect(req.Header.Get("display_name")).To(Equal("testhost")) + + var plugInfo []pluginDetail + plInfo := []byte(req.Header.Get("plugin_details")) + err = json.Unmarshal(plInfo, &plugInfo) + Expect(err).NotTo(HaveOccurred()) + + Expect(plugInfo[0].Name).To(Equal("apidApigeeSync")) + Expect(plugInfo[0].SchemaVersion).To(Equal("0.0.2")) + + res := oauthTokenResp{} + res.AccessToken = "accesstoken" + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + w.Write(body) + + }).Methods("POST") + + testRouter.HandleFunc("/snapshots", func(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + + q := req.URL.Query() + + if phase == 0 { + phase = 1 + Expect(q.Get("scope")).To(Equal(testScope)) + Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) + + apidcfgItem := common.Row{} + apidcfgItems := []common.Row{} + apidcfgItemCh := common.Row{} + apidcfgItemsCh := []common.Row{} + scv := &common.ColumnVal{ + Value: testScope, + Type: 1, + } + apidcfgItem["id"] = scv + scv = &common.ColumnVal{ + Value: testScope, + Type: 1, + } + apidcfgItem["_change_selector"] = scv + apidcfgItems = append(apidcfgItems, apidcfgItem) + + scv = &common.ColumnVal{ + Value: "apid_config_scope_id_0", + Type: 1, + } + apidcfgItemCh["id"] = scv + + scv = &common.ColumnVal{ + Value: "apid_config_scope_id_0", + Type: 1, + } + apidcfgItemCh["_change_selector"] = scv + + scv = &common.ColumnVal{ + Value: testScope, + Type: 1, + } + apidcfgItemCh["apid_cluster_id"] = scv + + scv = &common.ColumnVal{ + Value: "ert452", + Type: 1, + } + apidcfgItemCh["scope"] = scv + + { + scv = &common.ColumnVal{ + Value: "att", + Type: 1, + } + apidcfgItemCh["org"] = scv + + } + { + scv = &common.ColumnVal{ + Value: "prod", + Type: 1, + } + apidcfgItemCh["env"] = scv + } + + apidcfgItemsCh = append(apidcfgItemsCh, apidcfgItemCh) + + res := &common.Snapshot{} + res.SnapshotInfo = "snapinfo1" + + res.Tables = []common.Table{ + { + Name: "edgex.apid_cluster", + Rows: apidcfgItems, + }, + { + Name: "edgex.data_scope", + Rows: apidcfgItemsCh, + }, + } + + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + + w.Write(body) + return + } else { + phase = 2 + Expect(q.Get("scope")).To(Equal("ert452")) + res := &common.Snapshot{} + res.SnapshotInfo = "snapinfo1" + + apidcfgItems := []common.Row{} + res.Tables = []common.Table{ + { + Name: "kms.api_product", + Rows: apidcfgItems, + }, + } + + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + + w.Write(body) + return + } + + }).Methods("GET") + + testRouter.HandleFunc("/changes", func(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + + if req.URL.Query().Get("since") == "lastSeq_01" { + go func() { + block, err := strconv.Atoi(req.URL.Query().Get("block")) + Expect(err).NotTo(HaveOccurred()) + time.Sleep(time.Duration(block) * time.Second) + w.WriteHeader(http.StatusNotModified) + }() + return + } + + Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) + q := req.URL.Query() + Expect(q.Get("snapshot")).To(Equal("snapinfo1")) + scparams := q["scope"] + Expect(scparams).To(ContainElement("ert452")) + Expect(scparams).To(ContainElement("bootstrap")) + + res := &common.ChangeList{} + + res.LastSequence = "lastSeq_01" + mpItems := common.Row{} + + scv := &common.ColumnVal{ + Value: "apid_config_scope_id_1", + Type: 1, + } + mpItems["id"] = scv + + scv = &common.ColumnVal{ + Value: testScope, + Type: 1, + } + mpItems["apid_cluster_id"] = scv + + scv = &common.ColumnVal{ + Value: "ert452", + Type: 1, + } + mpItems["scope"] = scv + { + scv = &common.ColumnVal{ + Value: "att", + Type: 1, + } + mpItems["org"] = scv + } + { + scv = &common.ColumnVal{ + Value: "prod", + Type: 1, + } + mpItems["env"] = scv + } + + res.Changes = []common.Change{ + { + Table: "edgex.data_scope", + NewRow: mpItems, + Operation: 1, + }, + } + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + w.Write(body) + + }).Methods("GET") + + apid.InitializePlugins() }) +var _ = BeforeEach(func() { + apid.Events().Close() + + token = "" + downloadDataSnapshot = false + downloadBootSnapshot = false + changeFinished = false + lastSequence = "" + + _, err := getDB().Exec("DELETE FROM APID_CLUSTER") + Expect(err).NotTo(HaveOccurred()) + _, err = getDB().Exec("DELETE FROM DATA_SCOPE") + Expect(err).NotTo(HaveOccurred()) + + db, err := data.DB() + Expect(err).NotTo(HaveOccurred()) + _, err = db.Exec("DELETE FROM APID") + Expect(err).NotTo(HaveOccurred()) +}) + + var _ = AfterSuite(func() { apid.Events().Close() if testServer != nil {
diff --git a/apigee_sync.go b/apigee_sync.go index 2ece2a0..1ffaa18 100644 --- a/apigee_sync.go +++ b/apigee_sync.go
@@ -103,7 +103,7 @@ * Check to see if we have lastSequence already saved in the DB, * in which case, it has to be used to prevent re-reading same data */ - lastSequence = findApidConfigInfo("last_sequence") + lastSequence = findApidConfigInfo(lastSequence) for { log.Debug("polling...") if token == "" { @@ -338,6 +338,8 @@ func downloadSnapshot() { + log.Debugf("downloadSnapshot") + var scopes []string /* Get the bearer token */
diff --git a/apigee_sync_test.go b/apigee_sync_test.go index d421e3b..660230d 100644 --- a/apigee_sync_test.go +++ b/apigee_sync_test.go
@@ -1,273 +1,51 @@ package apidApigeeSync -import ( - "encoding/json" - "github.com/30x/apid" - "github.com/apigee-labs/transicator/common" - . "github.com/onsi/ginkgo" - . "github.com/onsi/gomega" - "net/http" - "time" -) - -var _ = Describe("api", func() { - - var phase = 0 - - It("should perform all sync phases", func(done Done) { - - // mock upstream testServer - testRouter.HandleFunc("/accesstoken", func(w http.ResponseWriter, req *http.Request) { - defer GinkgoRecover() - - Expect(req.Method).To(Equal("POST")) - Expect(req.Header.Get("Content-Type")).To(Equal("application/x-www-form-urlencoded; param=value")) - - err := req.ParseForm() - Expect(err).NotTo(HaveOccurred()) - Expect(req.Form.Get("grant_type")).To(Equal("client_credentials")) - Expect(req.Header.Get("status")).To(Equal("ONLINE")) - Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) - Expect(req.Header.Get("display_name")).To(Equal("testhost")) - - var plugInfo []pluginDetail - plInfo := []byte(req.Header.Get("plugin_details")) - err = json.Unmarshal(plInfo, &plugInfo) - Expect(err).NotTo(HaveOccurred()) - - Expect(plugInfo[0].Name).To(Equal("apidApigeeSync")) - Expect(plugInfo[0].SchemaVersion).To(Equal("0.0.2")) - - res := oauthTokenResp{} - res.AccessToken = "accesstoken" - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - w.Write(body) - - }).Methods("POST") - - testRouter.HandleFunc("/snapshots", func(w http.ResponseWriter, req *http.Request) { - defer GinkgoRecover() - - q := req.URL.Query() - - if phase == 0 { - phase = 1 - Expect(q.Get("scope")).To(Equal(testScope)) - Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) - - apidcfgItem := common.Row{} - apidcfgItems := []common.Row{} - apidcfgItemCh := common.Row{} - apidcfgItemsCh := []common.Row{} - scv := &common.ColumnVal{ - Value: testScope, - Type: 1, - } - apidcfgItem["id"] = scv - scv = &common.ColumnVal{ - Value: testScope, - Type: 1, - } - apidcfgItem["_change_selector"] = scv - apidcfgItems = append(apidcfgItems, apidcfgItem) - - scv = &common.ColumnVal{ - Value: "apid_config_scope_id_0", - Type: 1, - } - apidcfgItemCh["id"] = scv - - scv = &common.ColumnVal{ - Value: "apid_config_scope_id_0", - Type: 1, - } - apidcfgItemCh["_change_selector"] = scv - - scv = &common.ColumnVal{ - Value: testScope, - Type: 1, - } - apidcfgItemCh["apid_cluster_id"] = scv - - scv = &common.ColumnVal{ - Value: "ert452", - Type: 1, - } - apidcfgItemCh["scope"] = scv - - { - scv = &common.ColumnVal{ - Value: "att", - Type: 1, - } - apidcfgItemCh["org"] = scv - - } - { - scv = &common.ColumnVal{ - Value: "prod", - Type: 1, - } - apidcfgItemCh["env"] = scv - } - - apidcfgItemsCh = append(apidcfgItemsCh, apidcfgItemCh) - - res := &common.Snapshot{} - res.SnapshotInfo = "snapinfo1" - - res.Tables = []common.Table{ - { - Name: "edgex.apid_cluster", - Rows: apidcfgItems, - }, - { - Name: "edgex.data_scope", - Rows: apidcfgItemsCh, - }, - } - - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - - log.Debugf("/snapshots writing: %v", body) - - w.Write(body) - return - } else { - phase = 2 - Expect(q.Get("scope")).To(Equal("ert452")) - res := &common.Snapshot{} - res.SnapshotInfo = "snapinfo1" - - apidcfgItems := []common.Row{} - res.Tables = []common.Table{ - { - Name: "kms.api_product", - Rows: apidcfgItems, - }, - } - - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - - log.Debugf("/snapshots writing: %v", body) - - w.Write(body) - return - } - - }).Methods("GET") - - testRouter.HandleFunc("/changes", func(w http.ResponseWriter, req *http.Request) { - defer GinkgoRecover() - - Expect(req.Header.Get("apid_cluster_Id")).To(Equal("bootstrap")) - q := req.URL.Query() - Expect(q.Get("snapshot")).To(Equal("snapinfo1")) - scparams := q["scope"] - Expect(scparams).To(ContainElement("ert452")) - Expect(scparams).To(ContainElement("bootstrap")) - - res := &common.ChangeList{} - - res.LastSequence = "lastSeq_01" - mpItems := common.Row{} - - scv := &common.ColumnVal{ - Value: "apid_config_scope_id_1", - Type: 1, - } - mpItems["id"] = scv - - scv = &common.ColumnVal{ - Value: testScope, - Type: 1, - } - mpItems["apid_cluster_id"] = scv - - scv = &common.ColumnVal{ - Value: "ert452", - Type: 1, - } - mpItems["scope"] = scv - { - scv = &common.ColumnVal{ - Value: "att", - Type: 1, - } - mpItems["org"] = scv - } - { - scv = &common.ColumnVal{ - Value: "prod", - Type: 1, - } - mpItems["env"] = scv - } - - res.Changes = []common.Change{ - { - Table: "edgex.data_scope", - NewRow: mpItems, - Operation: 1, - }, - } - body, err := json.Marshal(res) - Expect(err).NotTo(HaveOccurred()) - w.Write(body) - - }).Methods("GET") - - apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) { - defer GinkgoRecover() - if _, ok := event.(*common.Snapshot); ok { - if phase > 1 { - db := getDB() - // verify event data (post snapshot) - var count int - err := db.QueryRow(` - Select count(scp.id) - FROM data_scope AS scp - INNER JOIN apid_cluster AS ap - WHERE scp.apid_cluster_id = ap.id - `).Scan(&count) - Expect(err).NotTo(HaveOccurred()) - Expect(count).Should(Equal(1)) - } - } else if _, ok := event.(*common.ChangeList); ok { - // verify event data (post change) - // There should be 2 scopes now - time.Sleep(200 * time.Millisecond) - db := getDB() - var count int - err := db.QueryRow(` - SELECT count(scp.id) - FROM data_scope AS scp - INNER JOIN apid_cluster AS ap - WHERE scp.apid_cluster_id = ap.id - `).Scan(&count) - Expect(err).NotTo(HaveOccurred()) - Expect(count).Should(Equal(2)) - close(done) - } else { - Fail("Unexpected event") - } - - }) - }) -}) - -type test_handler struct { - description string - f func(event apid.Event) -} - -func (t *test_handler) String() string { - return t.description -} - -func (t *test_handler) Handle(event apid.Event) { - t.f(event) -} +//import ( +// "github.com/30x/apid" +// "github.com/apigee-labs/transicator/common" +// . "github.com/onsi/ginkgo" +// . "github.com/onsi/gomega" +//) +// +//var _ = Describe("api", func() { +// +// It("should perform all sync phases", func(done Done) { +// +// apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) { +// defer GinkgoRecover() +// if _, ok := event.(*common.Snapshot); ok { +// if phase > 1 { +// db := getDB() +// // verify event data (post snapshot) +// var count int +// err := db.QueryRow(` +// Select count(scp.id) +// FROM data_scope AS scp +// INNER JOIN apid_cluster AS ap +// WHERE scp.apid_cluster_id = ap.id +// `).Scan(&count) +// Expect(err).NotTo(HaveOccurred()) +// Expect(count).To(Equal(1)) +// } +// } else if _, ok := event.(*common.ChangeList); ok { +// // verify event data (post change) +// // There should be 2 scopes now +// //time.Sleep(200 * time.Millisecond) +// db := getDB() +// var count int +// err := db.QueryRow(` +// SELECT count(scp.id) +// FROM data_scope AS scp +// INNER JOIN apid_cluster AS ap +// WHERE scp.apid_cluster_id = ap.id +// `).Scan(&count) +// Expect(err).NotTo(HaveOccurred()) +// Expect(count).To(Equal(2)) +// close(done) +// } else { +// Fail("Unexpected event") +// } +// +// }) +// }) +//})
diff --git a/data.go b/data.go index 6d85f04..ee4de00 100644 --- a/data.go +++ b/data.go
@@ -7,6 +7,7 @@ "time" "fmt" "crypto/rand" + "errors" ) var ( @@ -15,13 +16,13 @@ ) type dataApidCluster struct { - Scope, ID, Name, OrgAppName, CreatedBy, UpdatedBy, Description string - Updated, Created int64 + ChangeSelector, ID, Name, OrgAppName, CreatedBy, UpdatedBy, Description string + Updated, Created string } type dataDataScope struct { - ID, ClusterID, Scope, Org, Env, CreatedBy, UpdatedBy string - Updated, Created int64 + ChangeSelector, ID, ClusterID, Scope, Org, Env, CreatedBy, UpdatedBy string + Updated, Created string } /* @@ -42,9 +43,9 @@ name text, description text, umbrella_org_app_name text, - created int64, + created text, created_by text, - updated int64, + updated text, updated_by text, _change_selector text, last_sequence text, @@ -56,12 +57,12 @@ scope text, org text, env text, - created int64, + created text, created_by text, - updated int64, + updated text, updated_by text, _change_selector text, - PRIMARY KEY (id) + PRIMARY KEY (id, apid_cluster_id) ); `) if err != nil { @@ -85,7 +86,6 @@ dbMux.Unlock() } - func insertApidCluster(dac dataApidCluster, txn *sql.Tx) error { log.Debugf("inserting into APID_CLUSTER: %v", dac) @@ -93,8 +93,9 @@ stmt, err := txn.Prepare(` INSERT INTO APID_CLUSTER (id, _change_selector, name, umbrella_org_app_name, - created, created_by, updated, updated_by) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8); + created, created_by, updated, updated_by, + description) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9); `) if err != nil { log.Errorf("prepare insert into APID_CLUSTER transaction Failed: %v", err) @@ -103,14 +104,9 @@ defer stmt.Close() _, err = stmt.Exec( - dac.ID, - dac.Scope, - dac.Name, - dac.OrgAppName, - dac.Created, - dac.CreatedBy, - dac.Updated, - dac.UpdatedBy) + dac.ID, dac.ChangeSelector, dac.Name, dac.OrgAppName, + dac.Created, dac.CreatedBy, dac.Updated, dac.UpdatedBy, + dac.Description) if err != nil { log.Errorf("insert APID_CLUSTER failed: %v", err) @@ -127,8 +123,8 @@ INSERT INTO DATA_SCOPE (id, apid_cluster_id, scope, org, env, created, created_by, updated, - updated_by) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9); + updated_by, _change_selector) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10); `) if err != nil { log.Errorf("insert DATA_SCOPE failed: %v", err) @@ -137,74 +133,33 @@ defer stmt.Close() _, err = stmt.Exec( - ds.ID, - ds.ClusterID, - ds.Scope, - ds.Org, - ds.Env, - ds.Created, - ds.CreatedBy, - ds.Updated, - ds.UpdatedBy) + ds.ID, ds.ClusterID, ds.Scope, ds.Org, + ds.Env, ds.Created, ds.CreatedBy, ds.Updated, + ds.UpdatedBy, ds.ChangeSelector) if err != nil { - log.Error("insert DATA_SCOPE failed: %v", err) + log.Errorf("insert DATA_SCOPE failed: %v", err) return err } return nil } -func updateDataScope(ds dataDataScope, txn *sql.Tx) error { +func deleteDataScope(ds dataDataScope, txn *sql.Tx) error { - log.Debug("update DATA_SCOPE: %v", ds) + log.Debugf("delete DATA_SCOPE: %v", ds) - stmt, err := txn.Prepare(` - UPDATE DATA_SCOPE - SET apid_cluster_id=$2, scope=$3, org=$4, env=$5, - created=$6, created_by=$7, updated=$8, updated_by=$9 - WHERE id=$1 - `) + stmt, err := txn.Prepare("DELETE FROM DATA_SCOPE WHERE id=$1 and apid_cluster_id=$2") if err != nil { log.Errorf("update DATA_SCOPE failed: %v", err) return err } defer stmt.Close() - _, err = stmt.Exec( - ds.ID, - ds.ClusterID, - ds.Scope, - ds.Org, - ds.Env, - ds.Created, - ds.CreatedBy, - ds.Updated, - ds.UpdatedBy) + _, err = stmt.Exec(ds.ID, ds.ClusterID) if err != nil { - log.Error("update DATA_SCOPE failed: %v", err) - return err - } - - return nil -} - -func deleteDataScope(id string, txn *sql.Tx) error { - - log.Debug("delete DATA_SCOPE: %v", id) - - stmt, err := txn.Prepare("DELETE FROM DATA_SCOPE WHERE id=$1") - if err != nil { - log.Errorf("update DATA_SCOPE failed: %v", err) - return err - } - defer stmt.Close() - - _, err = stmt.Exec(id) - - if err != nil { - log.Error("delete DATA_SCOPE failed: %v", err) + log.Errorf("delete DATA_SCOPE failed: %v", err) return err } @@ -273,12 +228,12 @@ txn, err := db.Begin() if err != nil { - log.Error("Unable to create Sqlite transaction") + log.Errorf("Unable to create Sqlite transaction: %v", err) return false } prep, err := txn.Prepare("UPDATE APID_CLUSTER SET last_sequence=$1;") if err != nil { - log.Error("UPDATE APID_CLUSTER Failed: ", err) + log.Errorf("UPDATE APID_CLUSTER Failed: %v", err) return false } defer prep.Close() @@ -286,7 +241,7 @@ _, err = s.Exec(lastChange) s.Close() if err != nil { - log.Error("UPDATE DATA_SCOPE Failed: ", err) + log.Errorf("UPDATE DATA_SCOPE Failed: %v", err) txn.Rollback() return false } @@ -305,17 +260,20 @@ return } - err = db.QueryRow("SELECT instance_id, last_snapshot_info FROM APID"). + err = db.QueryRow("SELECT instance_id, last_snapshot_info FROM APID LIMIT 1"). Scan(&info.InstanceID, &info.LastSnapshot) - if err != nil && err != sql.ErrNoRows { - log.Errorf("Unable to retrieve apidInstanceInfo: %v", err) - return - } else { - // first start - no row, generate a UUID and store it - err = nil - info.InstanceID = generateUUID() + if err != nil { + if err != sql.ErrNoRows { + log.Errorf("Unable to retrieve apidInstanceInfo: %v", err) + return + } else { + log.Print("*** err: ", err) + // first start - no row, generate a UUID and store it + err = nil + info.InstanceID = generateUUID() - db.Exec("INSERT INTO APID (instance_id) VALUES (?)", info.InstanceID) + db.Exec("INSERT INTO APID (instance_id) VALUES (?)", info.InstanceID) + } } // if name not explicitly configured, just use InstanceID @@ -336,8 +294,18 @@ return err } - _, err = db.Exec("UPDATE APID SET last_snapshot_info=? WHERE instance_id=?", - apidInfo.LastSnapshot, apidInfo.InstanceID) + rows, err := db.Exec(` + INSERT OR REPLACE + INTO APID (instance_id, last_snapshot_info) + VALUES (?, ?)`, + apidInfo.InstanceID, apidInfo.LastSnapshot) + if err != nil { + return err + } + n, err := rows.RowsAffected() + if err == nil && n == 0 { + err = errors.New("no rows affected") + } return err }
diff --git a/listener.go b/listener.go index 524dc16..41b7a8e 100644 --- a/listener.go +++ b/listener.go
@@ -5,6 +5,11 @@ "github.com/apigee-labs/transicator/common" ) +const ( + LISTENER_TABLE_APID_CLUSTER = "edgex.apid_cluster" + LISTENER_TABLE_DATA_SCOPE = "edgex.data_scope" +) + type handler struct { } @@ -46,7 +51,7 @@ for _, table := range snapshot.Tables { switch table.Name { - case "edgex.apid_cluster": + case LISTENER_TABLE_APID_CLUSTER: if len(table.Rows) != 1 { log.Panic("Illegal state for apid_cluster. Must be a single row.") } @@ -56,7 +61,7 @@ log.Panic("Snapshot update failed: %v", err) } - case "edgex.data_scope": + case LISTENER_TABLE_DATA_SCOPE: for _, row := range table.Rows { ds := makeDataScopeFromRow(row) err := insertDataScope(ds, tx) @@ -79,7 +84,7 @@ } setDB(db) - log.Debug("Snapshot processed") + log.Debugf("Snapshot processed: %s", snapshot.SnapshotInfo) } func processChangeList(changes *common.ChangeList) { @@ -108,9 +113,8 @@ ds := makeDataScopeFromRow(change.NewRow) err = insertDataScope(ds, tx) case common.Delete: - var id string - change.OldRow.Get("ID", &id) - deleteDataScope(id, tx) + ds := makeDataScopeFromRow(change.OldRow) + deleteDataScope(ds, tx) default: // common.Update is not allowed log.Panicf("illegal operation: %s for %s", change.Operation, change.Table) @@ -132,7 +136,7 @@ dac := dataApidCluster{} row.Get("id", &dac.ID) - row.Get("_change_selector", &dac.Scope) + row.Get("_change_selector", &dac.ChangeSelector) row.Get("name", &dac.Name) row.Get("umbrella_org_app_name", &dac.OrgAppName) row.Get("created", &dac.Created) @@ -149,7 +153,7 @@ ds := dataDataScope{} row.Get("id", &ds.ID) - row.Get("_change_selector", &ds.Scope) + row.Get("_change_selector", &ds.ChangeSelector) row.Get("apid_cluster_id", &ds.ClusterID) row.Get("scope", &ds.Scope) row.Get("org", &ds.Org)
diff --git a/listener_test.go b/listener_test.go index 03c8ae8..02d255d 100644 --- a/listener_test.go +++ b/listener_test.go
@@ -1,2 +1,338 @@ package apidApigeeSync +import ( + . "github.com/onsi/ginkgo" + . "github.com/onsi/gomega" + + "github.com/apigee-labs/transicator/common" +) + +var _ = Describe("listener", func() { + + handler := handler{} + + Context("ApigeeSync snapshot event", func() { + + It("should set DB to appropriate version", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot", + Tables: []common.Table{}, + } + + handler.Handle(&event) + + Expect(apidInfo.LastSnapshot).To(Equal(event.SnapshotInfo)) + + expectedDB, err := data.DBVersion(event.SnapshotInfo) + Expect(err).NotTo(HaveOccurred()) + + Expect(getDB() == expectedDB).Should(BeTrue()) + }) + + It("should fail if zero apid_cluster rows", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot_fail", + Tables: []common.Table{ + { + Name: LISTENER_TABLE_APID_CLUSTER, + Rows: []common.Row{}, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + It("should fail if more than one apid_cluster rows", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot_fail", + Tables: []common.Table{ + { + Name: LISTENER_TABLE_APID_CLUSTER, + Rows: []common.Row{{}, {}}, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + It("should process a valid Snapshot", func() { + + event := common.Snapshot{ + SnapshotInfo: "test_snapshot_valid", + Tables: []common.Table{ + { + Name: LISTENER_TABLE_APID_CLUSTER, + Rows: []common.Row{ + { + "id": &common.ColumnVal{Value: "i"}, + "_change_selector": &common.ColumnVal{Value: "c"}, + "name": &common.ColumnVal{Value: "n"}, + "umbrella_org_app_name": &common.ColumnVal{Value: "o"}, + "created": &common.ColumnVal{Value: "c"}, + "created_by": &common.ColumnVal{Value: "c"}, + "updated": &common.ColumnVal{Value: "u"}, + "updated_by": &common.ColumnVal{Value: "u"}, + "description": &common.ColumnVal{Value: "d"}, + }, + }, + }, + { + Name: LISTENER_TABLE_DATA_SCOPE, + Rows: []common.Row{ + { + "id": &common.ColumnVal{Value: "i"}, + "_change_selector": &common.ColumnVal{Value: "c"}, + "apid_cluster_id": &common.ColumnVal{Value: "a"}, + "scope": &common.ColumnVal{Value: "s"}, + "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"}, + }, + }, + }, + }, + } + + handler.Handle(&event) + + info, err := getApidInstanceInfo() + Expect(err).NotTo(HaveOccurred()) + + Expect(info.LastSnapshot).To(Equal(event.SnapshotInfo)) + + db := getDB() + + // apid Cluster + var dcs []dataApidCluster + + rows, err := db.Query(` + SELECT id, name, description, umbrella_org_app_name, + created, created_by, updated, updated_by, + _change_selector + FROM 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, + &c.ChangeSelector) + 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")) + Expect(dc.ChangeSelector).To(Equal("c")) + + // Data Scope + var dds []dataDataScope + + rows, err = db.Query(` + SELECT id, apid_cluster_id, scope, org, + env, created, created_by, updated, + updated_by, _change_selector + FROM 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, &d.ChangeSelector) + dds = append(dds, d) + } + + Expect(len(dds)).To(Equal(1)) + 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("s")) + Expect(ds.Created).To(Equal("c")) + Expect(ds.CreatedBy).To(Equal("c")) + Expect(ds.Updated).To(Equal("u")) + Expect(ds.UpdatedBy).To(Equal("u")) + Expect(ds.ChangeSelector).To(Equal("c")) + }) + }) + + Context("ApigeeSync change event", func() { + + Context(LISTENER_TABLE_APID_CLUSTER, func() { + + It("insert event should panic", func() { + + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Insert, + Table: LISTENER_TABLE_APID_CLUSTER, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + It("update event should panic", func() { + + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Update, + Table: LISTENER_TABLE_APID_CLUSTER, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + PIt("delete event should kill all the things!") + }) + + Context(LISTENER_TABLE_DATA_SCOPE, func() { + + It("insert event should add", func() { + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Insert, + Table: LISTENER_TABLE_DATA_SCOPE, + NewRow: common.Row{ + "id": &common.ColumnVal{Value: "i"}, + "_change_selector": &common.ColumnVal{Value: "c"}, + "apid_cluster_id": &common.ColumnVal{Value: "a"}, + "scope": &common.ColumnVal{Value: "s"}, + "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"}, + }, + }, + }, + } + + handler.Handle(&event) + + var dds []dataDataScope + + rows, err := getDB().Query(` + SELECT id, apid_cluster_id, scope, org, + env, created, created_by, updated, + updated_by, _change_selector + FROM 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, &d.ChangeSelector) + dds = append(dds, d) + } + + Expect(len(dds)).To(Equal(1)) + 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("s")) + Expect(ds.Created).To(Equal("c")) + Expect(ds.CreatedBy).To(Equal("c")) + Expect(ds.Updated).To(Equal("u")) + Expect(ds.UpdatedBy).To(Equal("u")) + Expect(ds.ChangeSelector).To(Equal("c")) + }) + + It("delete event should delete", func() { + insert := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Insert, + Table: LISTENER_TABLE_DATA_SCOPE, + NewRow: common.Row{ + "id": &common.ColumnVal{Value: "i"}, + "_change_selector": &common.ColumnVal{Value: "c"}, + "apid_cluster_id": &common.ColumnVal{Value: "a"}, + "scope": &common.ColumnVal{Value: "s"}, + "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"}, + }, + }, + }, + } + + handler.Handle(&insert) + + delete := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Delete, + Table: LISTENER_TABLE_DATA_SCOPE, + OldRow: insert.Changes[0].NewRow, + }, + }, + } + + handler.Handle(&delete) + + var nRows int + err := getDB().QueryRow("SELECT count(id) FROM DATA_SCOPE").Scan(&nRows) + Expect(err).NotTo(HaveOccurred()) + + Expect(nRows).To(Equal(0)) + }) + + It("update event should panic", func() { + + event := common.ChangeList{ + LastSequence: "test", + Changes: []common.Change{ + { + Operation: common.Update, + Table: LISTENER_TABLE_DATA_SCOPE, + }, + }, + } + + Expect(func() { handler.Handle(&event) }).To(Panic()) + }) + + }) + + }) +})