Add a full-fledged mock server for load and perf testing
diff --git a/.gitignore b/.gitignore index 1e2283b..ac08ee2 100644 --- a/.gitignore +++ b/.gitignore
@@ -4,4 +4,5 @@ vendor *.iml .idea -*.lock \ No newline at end of file +*.lock +cmd/mockServer/mockServer
diff --git a/apigeeSync_suite_test.go b/apigeeSync_suite_test.go index 30f8dcb..476919b 100644 --- a/apigeeSync_suite_test.go +++ b/apigeeSync_suite_test.go
@@ -4,17 +4,14 @@ . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" - "encoding/json" "github.com/30x/apid" "github.com/30x/apid/factory" - "github.com/apigee-labs/transicator/common" "io/ioutil" - "net/http" "net/http/httptest" "os" - "strconv" "testing" "time" + "github.com/apigee-labs/transicator/common" ) var ( @@ -23,14 +20,10 @@ testRouter apid.Router ) -const testScope = "bootstrap" - var _ = BeforeSuite(func(done Done) { - var phase int - apid.Initialize(factory.DefaultServicesFactory()) - config := apid.Config() + config = apid.Config() var err error tmpDir, err = ioutil.TempDir("", "api_test") @@ -43,250 +36,21 @@ config.Set(configProxyServerBaseURI, testServer.URL) config.Set(configSnapServerBaseURI, testServer.URL) config.Set(configChangeServerBaseURI, testServer.URL) - config.Set(configApidClusterId, "apid_config_scope_0") - config.Set(configName, "testhost") - config.Set(configSnapshotProtocol, "json") - config.Set(configApidClusterId, testScope) + + config.Set(configName, "testhost") + config.Set(configApidClusterId, "bootstrap") config.Set(configConsumerKey, "XXXXXXX") config.Set(configConsumerSecret, "YYYYYYY") - // fake an unreliable api server - always fails the first time - fail := false - - // mock upstream testServer - testRouter.HandleFunc("/accesstoken", func(w http.ResponseWriter, req *http.Request) { - // make unreliable - fail = !fail - if fail { - w.WriteHeader(500) - return - } - - 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) { - // make unreliable - fail = !fail - if fail { - w.WriteHeader(500) - return - } - - 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 - scopes := q["scope"] - Expect(len(scopes)).Should(Equal(2)) - Expect(scopes).To(ContainElement(testScope)) - Expect(scopes).To(ContainElement("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) { - fail = !fail - if fail { - w.WriteHeader(500) - return - } - - 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")) - scope := q["scope"] - Expect(scope).To(ContainElement("ert452")) - Expect(scope).To(ContainElement(testScope)) - - 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") + registerMockServer(testRouter) // This is actually the first test :) // Tests that entire bootstrap and set of sync operations work apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) { - if s, ok := event.(*common.Snapshot); ok { + defer GinkgoRecover() - Expect(s.SnapshotInfo).Should(Equal("snapinfo1")) + if s, ok := event.(*common.Snapshot); ok { for _, t := range s.Tables { switch t.Name { @@ -302,8 +66,8 @@ Expect(id).To(Equal("bootstrap")) case "edgex.data_scope": - Expect(t.Rows).To(HaveLen(1)) - r := t.Rows[0] + Expect(t.Rows).To(HaveLen(2)) + r := t.Rows[1] // get the non-cluster row var cs, id, clusterID, env, org, scope string r.Get("_change_selector", &cs) @@ -313,24 +77,24 @@ r.Get("org", &org) r.Get("scope", &scope) - Expect(id).To(Equal("apid_config_scope_id_0")) - Expect(cs).To(Equal("apid_config_scope_id_0")) + Expect(id).To(Equal("ert452")) + Expect(cs).To(Equal("ert452")) + Expect(scope).To(Equal("ert452")) Expect(clusterID).To(Equal("bootstrap")) Expect(env).To(Equal("prod")) Expect(org).To(Equal("att")) - Expect(scope).To(Equal("ert452")) - case "kms.api_product": - Expect(t.Rows).To(HaveLen(0)) + //case "kms.api_product": + // Expect(t.Rows).To(HaveLen(0)) - default: - Fail("invalid table: " + t.Name) + //default: + // Fail("invalid table: " + t.Name) } } } else if cl, ok := event.(*common.ChangeList); ok { - Expect(cl.LastSequence).To(Equal("lastSeq_01")) + //Expect(cl.LastSequence).To(Equal("lastSeq_01")) Expect(cl.Changes).To(HaveLen(1)) c := cl.Changes[0] @@ -363,7 +127,7 @@ Scan(&seq) Expect(err).NotTo(HaveOccurred()) - Expect(seq).To(Equal("lastSeq_01")) + Expect(seq).To(Equal(cl.LastSequence)) close(done) })
diff --git a/apigee_sync.go b/apigee_sync.go index 3d59ddd..3ad929c 100644 --- a/apigee_sync.go +++ b/apigee_sync.go
@@ -147,6 +147,8 @@ return err } + // todo: should StatusNotChanged be a special case here? + /* If the call is not Authorized, update flag */ if r.StatusCode != http.StatusOK { if r.StatusCode == http.StatusUnauthorized { @@ -318,7 +320,7 @@ func Redirect(req *http.Request, via []*http.Request) error { req.Header.Add("Authorization", "Bearer "+token) - req.Header.Add("org", apidInfo.ClusterID) + req.Header.Add("org", apidInfo.ClusterID) // todo: this is strange.. is it needed? return nil }
diff --git a/cmd/mockServer/main.go b/cmd/mockServer/main.go new file mode 100644 index 0000000..e3cf6c3 --- /dev/null +++ b/cmd/mockServer/main.go
@@ -0,0 +1,52 @@ +package main + +import ( + "github.com/30x/apid" + "github.com/30x/apid/factory" + "time" + "github.com/30x/apidApigeeSync" +) + +func main() { + apid.Initialize(factory.DefaultServicesFactory()) + + log := apid.Log() + log.Debug("initializing...") + + config := apid.Config() + config.SetDefault("api_port", "9001") + + router := apid.API().Router() + //server := httptest.NewServer(testRouter) + + params := apidApigeeSync.MockParms{ + ReliableAPI: true, + ClusterID: "ZZZ", + TokenKey: "XXX", + TokenSecret: "YYY", + Scope: "ert452", + Organization: "att", + Environment: "prod", + NumDevelopers: 5, + NumApplicationsPerDeveloper: 1, + AddDeveloperEvery: 1 * time.Second, + UpdateDeveloperEvery: 1 * time.Second, + NumDeployments: 100, + ReplaceDeploymentEvery: 3 * time.Second, + } + + apidApigeeSync.Mock(params, router) + + // print the base url to the console + port := config.GetString("api_port") + log.Print() + log.Printf("API is at: http://localhost:%s", port) + log.Print() + + // start client API listener + api := apid.API() + err := api.Listen() + if err != nil { + log.Print(err) + } +}
diff --git a/listener.go b/listener.go index 2a95d6f..f488e68 100644 --- a/listener.go +++ b/listener.go
@@ -59,7 +59,7 @@ ac := makeApidClusterFromRow(row) err := insertApidCluster(ac, tx) if err != nil { - log.Panic("Snapshot update failed: %v", err) + log.Panicf("Snapshot update failed: %v", err) } } @@ -68,7 +68,7 @@ ds := makeDataScopeFromRow(row) err := insertDataScope(ds, tx) if err != nil { - log.Panic("Snapshot update failed: %v", err) + log.Panicf("Snapshot update failed: %v", err) } } }
diff --git a/mock_server.go b/mock_server.go new file mode 100644 index 0000000..72da244 --- /dev/null +++ b/mock_server.go
@@ -0,0 +1,527 @@ +package apidApigeeSync + +import ( + "encoding/json" + "github.com/30x/apid" + "github.com/apigee-labs/transicator/common" + . "github.com/onsi/gomega" + "net/http" + "strconv" + "time" + "fmt" + . "github.com/onsi/ginkgo" +) +/* +Currently limited to: 1 cluster, 1 data_scope, 1 organization, 1 env, 1 company + +Notes: +cluster_id is static per params +_change_selector == data_scope +data_scope == org + env +tenant_id == org + +company => * developer +developer => * app +application => * app_credential +product => * app_credential +*/ + +type MockParms struct { + ReliableAPI bool + ClusterID string + TokenKey string + TokenSecret string + Scope string + Organization string + Environment string + NumDevelopers int + NumApplicationsPerDeveloper int + AddDeveloperEvery time.Duration + UpdateDeveloperEvery time.Duration + NumDeployments int + ReplaceDeploymentEvery time.Duration +} + +func Mock(params MockParms, router apid.Router) *MockServer { + m := &MockServer{} + m.params = params + + m.init() + m.registerRoutes(router) + return m +} + +func registerMockServer(testRouter apid.Router) { + + p := MockParms{ + ReliableAPI: true, + ClusterID: config.GetString(configApidClusterId), + TokenKey: config.GetString(configConsumerKey), + TokenSecret: config.GetString(configConsumerSecret), + Scope: "ert452", + Organization: "att", + Environment: "prod", + NumDevelopers: 5, + NumApplicationsPerDeveloper: 1, + UpdateDeveloperEvery: 1 * time.Second, + NumDeployments: 100, + ReplaceDeploymentEvery: 3 * time.Second, + } + + m := MockServer{} + m.params = p + + m.init() + m.registerRoutes(testRouter) +} + +// table name -> common.Row +type tableRowMap map[string]common.Row + +type MockServer struct { + params MockParms + oauthToken string + snapshotID string + snapshotTables map[string][]common.Table // key = scopeID + sequenceID string + changeChannel chan []byte +} + +func (m *MockServer) init() { + m.changeChannel = make(chan []byte) + go m.developerGenerator() + + // cluster "scope" + cluster := m.newRow(map[string]string{ + "id": m.params.ClusterID, + "_change_selector": m.params.ClusterID, + }) + + // data scopes + var dataScopes []common.Row + dataScopes = append(dataScopes, cluster) + dataScopes = append(dataScopes, m.newRow(map[string]string{ + "id": m.params.Scope, + "scope": m.params.Scope, + "org": m.params.Organization, + "env": m.params.Environment, + "apid_cluster_id": m.params.ClusterID, + "_change_selector": m.params.Scope, + })) + + // cluster & data_scope snapshot tables + m.snapshotTables = map[string][]common.Table{} + m.snapshotTables[m.params.ClusterID] = []common.Table{ + { + Name: "edgex.apid_cluster", + Rows: []common.Row{cluster}, + }, + { + Name: "edgex.data_scope", + Rows: dataScopes, + }, + } + + var snapshotTableRows []tableRowMap + + // generate one company + companyID := m.params.Organization + tenantID := m.params.Organization + changeSelector := m.params.Scope + company := tableRowMap{ + "kms.company": m.newRow(map[string]string{ + "id": companyID, + "status": "Active", + "tenant_id": tenantID, + "_change_selector": changeSelector, + "name": companyID, + "display_name": companyID, + }), + } + snapshotTableRows = append(snapshotTableRows, company) + + // generate a bunch of developers + for i := 0; i < m.params.NumDevelopers; i++ { + developerID := generateUUID() + devRows := m.createDeveloper(developerID) + productID := generateUUID() + productRows := m.createProduct(productID) + appRows := m.createApplication(developerID, productID) + + developer := m.mergeTableRowMaps(devRows, productRows, appRows) + snapshotTableRows = append(snapshotTableRows, developer) + } + + m.snapshotTables[m.params.Scope] = m.concatTableRowMaps(snapshotTableRows...) +} + +func (m *MockServer) registerRoutes(router apid.Router) { + + router.HandleFunc("/accesstoken", m.unreliable(m.sendToken)).Methods("POST") + router.HandleFunc("/snapshots", m.unreliable(m.auth(m.sendSnapshot))).Methods("GET") + router.HandleFunc("/changes", m.unreliable(m.auth(m.sendChanges))).Methods("GET") +} + +func (m *MockServer) sendToken(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + m.registerFailHandler(w) + + 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(m.params.ClusterID)) + Expect(req.Header.Get("display_name")).ToNot(BeEmpty()) + + Expect(req.Form.Get("client_id")).To(Equal(m.params.TokenKey)) + Expect(req.Form.Get("client_secret")).To(Equal(m.params.TokenSecret)) + + 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).NotTo(BeEmpty()) + + m.oauthToken = generateUUID() + res := oauthTokenResp{ + AccessToken: m.oauthToken, + } + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + w.Write(body) +} + +func (m *MockServer) sendSnapshot(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + m.registerFailHandler(w) + + q := req.URL.Query() + scopes := q["scope"] + + Expect(scopes).To(ContainElement(m.params.ClusterID)) + + m.snapshotID = generateUUID() + snapshot := &common.Snapshot{ + SnapshotInfo: m.snapshotID, + } + + // Note: if/when we support multiple scopes, we'd have to do a merge of table rows + for _, scope := range scopes { + tables := m.snapshotTables[scope] + for _, table := range tables { + snapshot.AddTables(table) + } + } + + body, err := json.Marshal(snapshot) + Expect(err).NotTo(HaveOccurred()) + + w.Write(body) +} + +// todo: does "since" have any value? +func (m *MockServer) sendChanges(w http.ResponseWriter, req *http.Request) { + defer GinkgoRecover() + m.registerFailHandler(w) + + q := req.URL.Query() + scopes := q["scope"] + block, err := strconv.Atoi(req.URL.Query().Get("block")) + Expect(err).NotTo(HaveOccurred()) + since := req.URL.Query().Get("since") + + Expect(req.Header.Get("apid_cluster_Id")).To(Equal(m.params.ClusterID)) + Expect(q.Get("snapshot")).To(Equal(m.snapshotID)) + + Expect(scopes).To(ContainElement(m.params.ClusterID)) + Expect(scopes).To(ContainElement(m.params.Scope)) + + if block > 0 && since == m.sequenceID && since != "" { + m.sendChange(w, time.Duration(block) * time.Second) + return + } + + // todo: This is just legacy for the existing test in apigeeSync_suite_test + m.sequenceID = generateUUID() + res := &common.ChangeList{ + LastSequence: m.sequenceID, + } + + apidDataScopeRow := m.newRow(map[string]string{ + "id": "apid_config_scope_id_1", // adding a new scope + "scope": m.params.Scope, + "org": m.params.Organization, + "env": m.params.Environment, + "apid_cluster_id": m.params.ClusterID, + }) + + res.Changes = []common.Change{ + { + Table: "edgex.data_scope", + Operation: common.Insert, + NewRow: apidDataScopeRow, + }, + } + + body, err := json.Marshal(res) + Expect(err).NotTo(HaveOccurred()) + + w.Write(body) +} + +func (m *MockServer) developerGenerator() { + + tick := time.Tick(m.params.AddDeveloperEvery) + for range tick { + + // generate a random developer w/ product and app + developerID := generateUUID() + devRows := m.createDeveloper(developerID) + productID := generateUUID() + productRows := m.createProduct(productID) + appRows := m.createApplication(developerID, productID) + + developer := m.mergeTableRowMaps(devRows, productRows, appRows) + changeList := m.createInsertChange(developer) + + m.sequenceID = generateUUID() + changeList.LastSequence = m.sequenceID + + body, err := json.Marshal(changeList) + if err != nil { + fmt.Printf("Error generating developer!\n%v\n", err) + } + + fmt.Println("adding developer") + fmt.Println(string(body)) + m.changeChannel <- body + } +} + +func (m *MockServer) sendChange(w http.ResponseWriter, timeout time.Duration) { + select { + case change := <-m.changeChannel: + fmt.Println("sending change to client") + w.Write(change) + case <-time.After(timeout): + fmt.Println("change request timeout") + w.WriteHeader(http.StatusNotModified) + } +} + +// enforces handler auth +func (m *MockServer) auth(target http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { + auth := req.Header.Get("Authorization") + + if auth != fmt.Sprintf("Bearer %s", m.oauthToken) { + w.WriteHeader(http.StatusBadRequest) + } else { + target(w, req) + } + } +} + +// make a handler unreliable +func (m *MockServer) unreliable(target http.HandlerFunc) http.HandlerFunc { + if m.params.ReliableAPI { + return target + } + + var fail bool + return func(w http.ResponseWriter, req *http.Request) { + fail = !fail + if fail { + w.WriteHeader(500) + } else { + target(w, req) + } + } +} + +func (m *MockServer) registerFailHandler(w http.ResponseWriter) { + RegisterFailHandler(func(message string, callerSkip ...int) { + w.WriteHeader(400) + w.Write([]byte(message)) + fmt.Printf("sending error: %#v\n", message) + panic(GINKGO_PANIC) + }) +} + + +func (m *MockServer) newRow(keyAndVals map[string]string) (row common.Row) { + + row = common.Row{} + for k, v := range keyAndVals { + row[k] = &common.ColumnVal{ + Value: v, + Type: 1, + } + } + return +} + +func (m *MockServer) createDeveloper(developerID string) tableRowMap { + + companyID := m.params.Organization + tenantID := m.params.Organization + changeSelector := m.params.Scope + + rows := tableRowMap{} + + rows["kms.developer"] = m.newRow(map[string]string{ + "id": developerID, + "status": "Active", + "tenant_id": tenantID, + "_change_selector": changeSelector, + }) + + // map developer onto to existing company + rows["kms.company_developer"] = m.newRow(map[string]string{ + "id": developerID, + "tenant_id": tenantID, + "_change_selector": changeSelector, + "company_id": companyID, + "developer_id": developerID, + }) + + return rows +} + +func (m *MockServer) createProduct(productID string) tableRowMap { + + tenantID := m.params.Organization + changeSelector := m.params.Scope + + rows := tableRowMap{} + rows["kms.product"] = m.newRow(map[string]string{ + "id": productID, + "api_resources": "{}", + "environments": "{Env_0, Env_1}", + "tenant_id": tenantID, + "_change_selector": changeSelector, + }) + return rows +} + +func (m *MockServer) createApplication(developerID, productID string) tableRowMap { + + tenantID := m.params.Organization + changeSelector := m.params.Scope + + rows := tableRowMap{} + + applicationID := generateUUID() + credentialID := generateUUID() + + rows["kms.app"] = m.newRow(map[string]string{ + "id": applicationID, + "developer_id": developerID, + "status": "Approved", + "tenant_id": tenantID, + "_change_selector": changeSelector, + }) + + rows["kms.app_credential"] = m.newRow(map[string]string{ + "id": credentialID, + "app_id": applicationID, + "tenant_id": tenantID, + "status": "Approved", + "_change_selector": changeSelector, + }) + + rows["kms.app_credential_apiproduct_mapper"] = m.newRow(map[string]string{ + "apiprdt_id": productID, + "app_id": applicationID, + "appcred_id": credentialID, + "status": "Approved", + "_change_selector": changeSelector, + "tenant_id": tenantID, + }) + + return rows +} + +func (m *MockServer) createInsertChange(newRows tableRowMap) common.ChangeList { + + var changeList = common.ChangeList{} + for table, row := range newRows { + change := common.Change{ + Table: table, + NewRow: row, + Operation: common.Insert, + } + + changeList.Changes = append(changeList.Changes, change) + } + return changeList +} + +func (m *MockServer) createDeleteChange(oldRows tableRowMap) common.ChangeList { + + var changeList = common.ChangeList{} + for table, row := range oldRows { + change := common.Change{ + Table: table, + OldRow: row, + Operation: common.Delete, + } + + changeList.Changes = append(changeList.Changes, change) + } + return changeList +} + +func (m *MockServer) createUpdateChange(oldRows, newRows tableRowMap) common.ChangeList { + + var changeList = common.ChangeList{} + for table, oldRow := range oldRows { + change := common.Change{ + Table: table, + OldRow: oldRow, + NewRow: newRows[table], + Operation: common.Update, + } + + changeList.Changes = append(changeList.Changes, change) + } + return changeList +} + +// create one tableRowMap from various tableRowMap - tables must be unique +func (m *MockServer) mergeTableRowMaps(maps ...tableRowMap) tableRowMap { + merged := tableRowMap{} + for _, m := range maps { + for name, row := range m { + if _, ok:= merged[name]; ok { + panic(fmt.Sprintf("bad merge. name: %#v, row: %#v", name, row)) + } + merged[name] = row + } + } + return merged +} + +// create []common.Table from array of tableRowMaps +func (m *MockServer) concatTableRowMaps(maps ...tableRowMap) []common.Table { + tableMap := map[string]*common.Table{} + for _, m := range maps { + for name, row := range m { + if _, ok:= tableMap[name]; !ok { + tableMap[name] = &common.Table{ + Name: name, + } + } + tableMap[name].AddRowstoTable(row) + } + } + result := []common.Table{} + for _, v := range tableMap { + result = append(result, *v) + } + return result +}