Merge pull request #15 from 30x/XAPID-641
Create a mock server for unit, perf, and load 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..e8deef3 100644
--- a/apigeeSync_suite_test.go
+++ b/apigeeSync_suite_test.go
@@ -4,17 +4,15 @@
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
- "encoding/json"
+ "io/ioutil"
+ "net/http/httptest"
+ "os"
+ "testing"
+ "time"
+
"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"
)
var (
@@ -23,14 +21,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 +37,36 @@
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
+ log = apid.Log()
- // 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")
+ // set up mock server
+ mockParms := MockParms{
+ ReliableAPI: true,
+ ClusterID: config.GetString(configApidClusterId),
+ TokenKey: config.GetString(configConsumerKey),
+ TokenSecret: config.GetString(configConsumerSecret),
+ Scope: "ert452",
+ Organization: "att",
+ Environment: "prod",
+ }
+ Mock(mockParms, testRouter)
// This is actually the first test :)
// Tests that entire bootstrap and set of sync operations work
+ var lastSnapshot *common.Snapshot
apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
+ defer GinkgoRecover()
+
if s, ok := event.(*common.Snapshot); ok {
- Expect(s.SnapshotInfo).Should(Equal("snapinfo1"))
+ lastSnapshot = s
for _, t := range s.Tables {
switch t.Name {
@@ -294,63 +74,55 @@
case "edgex.apid_cluster":
Expect(t.Rows).To(HaveLen(1))
r := t.Rows[0]
- var cs, id string
- r.Get("_change_selector", &cs)
+ var id string
r.Get("id", &id)
-
- Expect(cs).To(Equal("bootstrap"))
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)
+ var id, clusterID, env, org, scope string
r.Get("id", &id)
r.Get("apid_cluster_id", &clusterID)
r.Get("env", &env)
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(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))
-
- default:
- Fail("invalid table: " + t.Name)
}
}
} else if cl, ok := event.(*common.ChangeList); ok {
- Expect(cl.LastSequence).To(Equal("lastSeq_01"))
- Expect(cl.Changes).To(HaveLen(1))
+ // ensure that snapshot switched DB versions
+ Expect(apidInfo.LastSnapshot).To(Equal(lastSnapshot.SnapshotInfo))
+ expectedDB, err := data.DBVersion(lastSnapshot.SnapshotInfo)
+ Expect(err).NotTo(HaveOccurred())
+ Expect(getDB() == expectedDB).Should(BeTrue())
- c := cl.Changes[0]
- Expect(c.Table).To(Equal("edgex.data_scope"))
- Expect(c.Operation).To(Equal(common.Insert))
+ Expect(cl.Changes).To(HaveLen(6))
- Expect(c.NewRow).ToNot(BeNil())
+ var tables []string
+ for _, c := range cl.Changes {
+ tables = append(tables, c.Table)
+ Expect(c.NewRow).ToNot(BeNil())
- var id, clusterID, env, org, scope string
- c.NewRow.Get("id", &id)
- c.NewRow.Get("apid_cluster_id", &clusterID)
- c.NewRow.Get("env", &env)
- c.NewRow.Get("org", &org)
- c.NewRow.Get("scope", &scope)
+ var tenantID string
+ c.NewRow.Get("tenant_id", &tenantID)
+ Expect(tenantID).To(Equal("ert452"))
+ }
- Expect(id).To(Equal("apid_config_scope_id_1"))
- Expect(clusterID).To(Equal("bootstrap"))
- Expect(env).To(Equal("prod"))
- Expect(org).To(Equal("att"))
- Expect(scope).To(Equal("ert452"))
+ Expect(tables).To(ContainElement("kms.app_credential"))
+ Expect(tables).To(ContainElement("kms.app_credential_apiproduct_mapper"))
+ Expect(tables).To(ContainElement("kms.developer"))
+ Expect(tables).To(ContainElement("kms.company_developer"))
+ Expect(tables).To(ContainElement("kms.api_product"))
+ Expect(tables).To(ContainElement("kms.app"))
events.ListenFunc(apid.EventDeliveredSelector, func(e apid.Event) {
defer GinkgoRecover()
@@ -363,7 +135,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..cecb838 100644
--- a/apigee_sync.go
+++ b/apigee_sync.go
@@ -4,13 +4,14 @@
"bytes"
"encoding/json"
"errors"
- "github.com/30x/apid"
- "github.com/apigee-labs/transicator/common"
"io/ioutil"
"net/http"
"net/url"
"path"
"time"
+
+ "github.com/30x/apid"
+ "github.com/apigee-labs/transicator/common"
)
var token string
@@ -31,7 +32,7 @@
if ev, ok := ede.Event.(*common.ChangeList); ok {
if lastSequence != ev.LastSequence {
lastSequence = ev.LastSequence
- err := persistChange(lastSequence)
+ err := updateLastSequence(lastSequence)
if err != nil {
log.Panic("Unable to update Sequence in DB")
}
@@ -105,7 +106,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(lastSequence)
+ lastSequence = getLastSequence()
for {
log.Debug("polling...")
if token == "" {
@@ -147,6 +148,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 {
@@ -166,14 +169,6 @@
return err
}
- if lastSequence != resp.LastSequence {
- lastSequence = resp.LastSequence
- err := persistChange(lastSequence)
- if err != nil {
- log.Panic("Unable to update Sequence in DB")
- }
- }
-
/* If valid data present, Emit to plugins */
if len(resp.Changes) > 0 {
changeFinished = false
@@ -201,7 +196,7 @@
if lastSequence != resp.LastSequence {
lastSequence = resp.LastSequence
- err := persistChange(lastSequence)
+ err := updateLastSequence(lastSequence)
if err != nil {
log.Panic("Unable to update Sequence in DB")
}
@@ -210,7 +205,6 @@
}
}
-
// simple doubling back-off
func createBackOff(retryIn, maxBackOff time.Duration) func() {
return func() {
@@ -318,7 +312,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
}
@@ -339,9 +333,6 @@
// Skip Downloading snapshot if there is already a snapshot available from previous run of APID
if apidInfo.LastSnapshot != "" {
- downloadDataSnapshot = true
- downloadBootSnapshot = true
-
log.Infof("Starting on downloaded snapshot: %s", apidInfo.LastSnapshot)
// ensure DB version will be accessible on behalf of dependant plugins
@@ -354,7 +345,12 @@
snap := &common.Snapshot{
SnapshotInfo: apidInfo.LastSnapshot,
}
- events.Emit(ApigeeSyncEventSelector, snap)
+ events.EmitWithCallback(ApigeeSyncEventSelector, snap, func(event apid.Event) {
+ downloadBootSnapshot = true
+ downloadDataSnapshot = true
+
+ go updatePeriodicChanges()
+ })
return
}
@@ -384,6 +380,8 @@
} else {
log.Panic("Snapshot for bootscope failed")
}
+
+ go updatePeriodicChanges()
}
func downloadSnapshot() {
diff --git a/cmd/mockServer/README.md b/cmd/mockServer/README.md
new file mode 100644
index 0000000..fd65ee5
--- /dev/null
+++ b/cmd/mockServer/README.md
@@ -0,0 +1,93 @@
+# apidApigeeSync Mock Server
+
+## Overview
+
+This Mock Server is used during unit tests of apidApigeeSync and has been designed to be run standalone
+for stand-alone development use as well as performance and load testing.
+
+## Build
+
+From the apidApigeeSync base dir:
+
+ glide install
+
+From apidApigeeSync/cmd/mockServer:
+
+ go build
+
+You should now have an executable named "mockServer".
+
+## Execute
+
+Execute with the -h flag to see flags:
+
+ ./mockServer -h
+ Usage of ./mockServer:
+ -addDevEach duration
+ add a developer each duration (default 0s)
+ -bundleURI string
+ a URI to a valid deployment bundle (default '')
+ -numDeps int
+ number of deployments in snapshot (default 2)
+ -numDevs int
+ number of developers in snapshot (default 2)
+ -reliable
+ if false, server will often send 500 errors (default true)
+ -upDepEach duration
+ update (replace) a deployment each duration (default 0s)
+ -upDevEach duration
+ update a developer each duration (default 0s)
+
+Note: Nothing is required.
+
+The following are the values used by default by the Mock Server:
+
+ ReliableAPI: true
+ ClusterID: "cluster"
+ TokenKey: "key"
+ TokenSecret: "secret"
+ Scope: "scope"
+ Organization: "org"
+ Environment: "test"
+ NumDevelopers: 2
+ AddDeveloperEvery: 0
+ UpdateDeveloperEvery: 0
+ NumDeployments: 2
+ ReplaceDeploymentEvery: 0
+ Port: 9001
+
+## Put it to use
+
+Set your apid configuration to point toward the Mock Server and have correct cluster, key, and secret values.
+
+For example:
+
+ api_port: 9000
+ api_expvar_path: /expvar
+ events_buffer_size: 5
+ log_level: debug
+ apigeesync_proxy_server_base: http://localhost:9001
+ apigeesync_snapshot_server_base: http://localhost:9001
+ apigeesync_change_server_base: http://localhost:9001
+ apigeesync_consumer_key: key
+ apigeesync_consumer_secret: secret
+ apigeesync_cluster_id: cluster
+ #data_trace_log_level: debug
+ data_source: file:%s?_busy_timeout=20000
+
+Now start apid. It should download the snapshot and changes as you configured for the Mock Server.
+
+Try out a couple of APIs to verify:
+
+ curl -i -d "action=verify&key=1&uriPath=/&scopeuuid=scope" :9000/verifiers/apikey
+
+ curl -i :9000/deployments
+
+## Notes
+
+Under high loads (eg. a large snapshot), apid may get timeout errors from sqlite.
+If you see this, you can work around it by increasing the _busy_timeout by adding a config item to your apid config:
+
+ data_source: file:%s?_busy_timeout=10000
+
+The _busy_timeout value is in milliseconds, so the above value is 10s.
diff --git a/cmd/mockServer/main.go b/cmd/mockServer/main.go
new file mode 100644
index 0000000..cab43f5
--- /dev/null
+++ b/cmd/mockServer/main.go
@@ -0,0 +1,76 @@
+package main
+
+import (
+ "flag"
+
+ "os"
+
+ "time"
+
+ "github.com/30x/apid"
+ "github.com/30x/apid/factory"
+ "github.com/30x/apidApigeeSync"
+)
+
+// runs a mock server standalone
+func main() {
+ // create new flag to avoid displaying all the Ginkgo flags
+ flag := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
+
+ bundleURI := *flag.String("bundleURI", "", "a URI to a valid deployment bundle (default '')")
+
+ reliable := *flag.Bool("reliable", true, "if false, server will often send 500 errors")
+
+ numDevs := *flag.Int("numDevs", 2, "number of developers in snapshot")
+ addDevEach := *flag.Duration("addDevEach", 0*time.Second, "add a developer each duration (default 0s)")
+ upDevEach := *flag.Duration("upDevEach", 0*time.Second, "update a developer each duration (default 0s)")
+
+ numDeps := *flag.Int("numDeps", 2, "number of deployments in snapshot")
+ upDepEach := *flag.Duration("upDepEach", 0*time.Second, "update (replace) a deployment each duration (default 0s)")
+
+ flag.Parse(os.Args[1:])
+
+ apid.Initialize(factory.DefaultServicesFactory())
+
+ log := apid.Log()
+ log.Debug("initializing...")
+ apidApigeeSync.SetLogger(log)
+
+ config := apid.Config()
+ config.SetDefault("api_port", "9001")
+
+ router := apid.API().Router()
+
+ params := apidApigeeSync.MockParms{
+ ReliableAPI: reliable,
+ ClusterID: "cluster",
+ TokenKey: "key",
+ TokenSecret: "secret",
+ Scope: "scope",
+ Organization: "org",
+ Environment: "test",
+ NumDevelopers: numDevs,
+ AddDeveloperEvery: addDevEach,
+ UpdateDeveloperEvery: upDevEach,
+ NumDeployments: numDeps,
+ ReplaceDeploymentEvery: upDepEach,
+ BundleURI: bundleURI,
+ }
+
+ log.Printf("Params: %#v\n", params)
+
+ 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/data.go b/data.go
index 6618531..3ce623e 100644
--- a/data.go
+++ b/data.go
@@ -15,12 +15,12 @@
)
type dataApidCluster struct {
- ChangeSelector, ID, Name, OrgAppName, CreatedBy, UpdatedBy, Description string
+ ID, Name, OrgAppName, CreatedBy, UpdatedBy, Description string
Updated, Created string
}
type dataDataScope struct {
- ChangeSelector, ID, ClusterID, Scope, Org, Env, CreatedBy, UpdatedBy string
+ ID, ClusterID, Scope, Org, Env, CreatedBy, UpdatedBy string
Updated, Created string
}
@@ -46,7 +46,6 @@
created_by text,
updated text,
updated_by text,
- _change_selector text,
last_sequence text,
PRIMARY KEY (id)
);
@@ -60,7 +59,6 @@
created_by text,
updated text,
updated_by text,
- _change_selector text,
PRIMARY KEY (id, apid_cluster_id)
);
`)
@@ -91,9 +89,9 @@
stmt, err := txn.Prepare(`
INSERT INTO APID_CLUSTER
- (id, _change_selector, name, umbrella_org_app_name,
+ (id, description, name, umbrella_org_app_name,
created, created_by, updated, updated_by,
- description)
+ last_sequence)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9);
`)
if err != nil {
@@ -103,9 +101,9 @@
defer stmt.Close()
_, err = stmt.Exec(
- dac.ID, dac.ChangeSelector, dac.Name, dac.OrgAppName,
+ dac.ID, dac.Description, 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)
@@ -122,8 +120,8 @@
INSERT INTO DATA_SCOPE
(id, apid_cluster_id, scope, org,
env, created, created_by, updated,
- updated_by, _change_selector)
- VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10);
+ updated_by)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9);
`)
if err != nil {
log.Errorf("insert DATA_SCOPE failed: %v", err)
@@ -134,7 +132,7 @@
_, err = stmt.Exec(
ds.ID, ds.ClusterID, ds.Scope, ds.Org,
ds.Env, ds.Created, ds.CreatedBy, ds.Updated,
- ds.UpdatedBy, ds.ChangeSelector)
+ ds.UpdatedBy)
if err != nil {
log.Errorf("insert DATA_SCOPE failed: %v", err)
@@ -194,24 +192,15 @@
/*
* Retrieve SnapshotInfo for the given apidConfigId from apid_config table
*/
-func findApidConfigInfo(qparam string) (info string) {
+func getLastSequence() (lastSequence string) {
- log.Debugf("findApidConfigInfo: %s", qparam)
-
- db := getDB()
-
- rows, err := db.Query("select ? from APID_CLUSTER", qparam)
- if err != nil {
- log.Errorf("Failed to query APID_CLUSTER: %v", err)
+ err := getDB().QueryRow("select last_sequence from APID_CLUSTER LIMIT 1").Scan(&lastSequence)
+ if err != nil && err != sql.ErrNoRows {
+ log.Panicf("Failed to query APID_CLUSTER: %v", err)
return
}
- defer rows.Close()
- for rows.Next() {
- rows.Scan(&info)
- }
- log.Debugf("info: %s", info)
-
+ log.Debugf("lastSequence: %s", lastSequence)
return
}
@@ -219,26 +208,24 @@
* Persist the last change Id each time a change has been successfully
* processed by the plugin(s)
*/
-func persistChange(lastChange string) error {
+func updateLastSequence(lastSequence string) error {
- log.Debugf("persistChange: %s", lastChange)
+ log.Debugf("updateLastSequence: %s", lastSequence)
- db := getDB()
-
- stmt, err := db.Prepare("UPDATE APID_CLUSTER SET last_sequence=$1;")
+ stmt, err := getDB().Prepare("UPDATE APID_CLUSTER SET last_sequence=$1;")
if err != nil {
log.Errorf("UPDATE APID_CLUSTER Failed: %v", err)
return err
}
defer stmt.Close()
- _, err = stmt.Exec(lastChange)
+ _, err = stmt.Exec(lastSequence)
if err != nil {
log.Errorf("UPDATE DATA_SCOPE Failed: %v", err)
return err
}
- log.Infof("UPDATE DATA_SCOPE Success: %s", lastChange)
+ log.Infof("UPDATE DATA_SCOPE Success: %s", lastSequence)
return nil
}
diff --git a/glide.yaml b/glide.yaml
index c07a74e..65d3e3d 100644
--- a/glide.yaml
+++ b/glide.yaml
@@ -2,7 +2,7 @@
import:
- package: github.com/30x/apid
version: master
-testImport:
- package: github.com/onsi/ginkgo/ginkgo
+ version: master
- package: github.com/onsi/gomega
version: master
diff --git a/init.go b/init.go
index e78b399..0b3d8c8 100644
--- a/init.go
+++ b/init.go
@@ -3,8 +3,9 @@
import (
"encoding/json"
"fmt"
- "github.com/30x/apid"
"os"
+
+ "github.com/30x/apid"
)
const (
@@ -50,6 +51,7 @@
func initDefaults() {
config.SetDefault(configPollInterval, 120)
+ config.SetDefault(configSnapshotProtocol, "json")
name, errh := os.Hostname()
if (errh != nil) && (len(config.GetString(configName)) == 0) {
log.Errorf("Not able to get hostname for kernel. Please set '%s' property in config", configName)
@@ -87,6 +89,10 @@
return pluginData, fmt.Errorf("Missing required config value: %s", key)
}
}
+ proto := config.GetString(configSnapshotProtocol)
+ if proto != "json" && proto != "proto" {
+ return pluginData, fmt.Errorf("Illegal value for %s. Must be: 'json' or 'proto'", configSnapshotProtocol)
+ }
// set up default database
db, err := data.DB()
@@ -147,10 +153,6 @@
go bootstrap()
- /* Begin Looking for changes periodically */
- log.Debug("starting update goroutine")
- go updatePeriodicChanges()
-
events.Listen(ApigeeSyncEventSelector, &handler{})
log.Debug("Done post plugin init")
}
diff --git a/listener.go b/listener.go
index 2a95d6f..bd4baee 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)
}
}
}
@@ -138,7 +138,6 @@
dac := dataApidCluster{}
row.Get("id", &dac.ID)
- row.Get("_change_selector", &dac.ChangeSelector)
row.Get("name", &dac.Name)
row.Get("umbrella_org_app_name", &dac.OrgAppName)
row.Get("created", &dac.Created)
@@ -155,7 +154,6 @@
ds := dataDataScope{}
row.Get("id", &ds.ID)
- 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 ad2c272..93a9588 100644
--- a/listener_test.go
+++ b/listener_test.go
@@ -55,7 +55,6 @@
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"},
@@ -71,7 +70,6 @@
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"},
@@ -100,8 +98,7 @@
rows, err := db.Query(`
SELECT id, name, description, umbrella_org_app_name,
- created, created_by, updated, updated_by,
- _change_selector
+ created, created_by, updated, updated_by
FROM APID_CLUSTER`)
Expect(err).NotTo(HaveOccurred())
defer rows.Close()
@@ -109,8 +106,7 @@
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)
+ &c.Created, &c.CreatedBy, &c.Updated, &c.UpdatedBy)
dcs = append(dcs, c)
}
@@ -125,7 +121,6 @@
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
@@ -133,7 +128,7 @@
rows, err = db.Query(`
SELECT id, apid_cluster_id, scope, org,
env, created, created_by, updated,
- updated_by, _change_selector
+ updated_by
FROM DATA_SCOPE`)
Expect(err).NotTo(HaveOccurred())
defer rows.Close()
@@ -142,7 +137,7 @@
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)
+ &d.UpdatedBy)
dds = append(dds, d)
}
@@ -157,7 +152,6 @@
Expect(ds.CreatedBy).To(Equal("c"))
Expect(ds.Updated).To(Equal("u"))
Expect(ds.UpdatedBy).To(Equal("u"))
- Expect(ds.ChangeSelector).To(Equal("c"))
})
})
@@ -209,7 +203,6 @@
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"},
@@ -230,7 +223,7 @@
rows, err := getDB().Query(`
SELECT id, apid_cluster_id, scope, org,
env, created, created_by, updated,
- updated_by, _change_selector
+ updated_by
FROM DATA_SCOPE`)
Expect(err).NotTo(HaveOccurred())
defer rows.Close()
@@ -239,7 +232,7 @@
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)
+ &d.UpdatedBy)
dds = append(dds, d)
}
@@ -254,7 +247,6 @@
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() {
@@ -266,7 +258,6 @@
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"},
diff --git a/mock_server.go b/mock_server.go
new file mode 100644
index 0000000..d8de52d
--- /dev/null
+++ b/mock_server.go
@@ -0,0 +1,672 @@
+package apidApigeeSync
+
+import (
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "hash/crc32"
+ "math/rand"
+ "net/http"
+ "net/url"
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/30x/apid"
+ "github.com/apigee-labs/transicator/common"
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+)
+
+/*
+Currently limited to:
+ 1 cluster, 1 scope, 1 org, 1 env, 1 company
+ 1 app & 1 product per developer
+
+Notes:
+ Scope ~= org + env
+ tenant_id == Scope for our purposes
+ (technically, data_scope.scope = tenant_id)
+
+Relations:
+ 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
+ AddDeveloperEvery time.Duration
+ UpdateDeveloperEvery time.Duration
+ NumDeployments int
+ ReplaceDeploymentEvery time.Duration
+ BundleURI string
+}
+
+func Mock(params MockParms, router apid.Router) *MockServer {
+ m := &MockServer{}
+ m.params = params
+
+ m.init()
+ m.registerRoutes(router)
+ return m
+}
+
+// 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
+ changeChannel chan []byte
+ sequenceID *int64
+ maxDevID *int64
+ deployIDMutex sync.RWMutex
+ minDeploymentID *int64
+ maxDeploymentID *int64
+}
+
+func (m *MockServer) lastSequenceID() string {
+ return strconv.FormatInt(atomic.LoadInt64(m.sequenceID), 10)
+}
+
+func (m *MockServer) nextSequenceID() string {
+ return strconv.FormatInt(atomic.AddInt64(m.sequenceID, 1), 10)
+}
+
+func (m *MockServer) nextDeveloperID() string {
+ return strconv.FormatInt(atomic.AddInt64(m.maxDevID, 1), 10)
+}
+
+func (m *MockServer) randomDeveloperID() string {
+ return strconv.FormatInt(rand.Int63n(atomic.LoadInt64(m.maxDevID)), 10)
+}
+
+func (m *MockServer) nextDeploymentID() string {
+ return strconv.FormatInt(atomic.AddInt64(m.maxDeploymentID, 1), 10)
+}
+
+func (m *MockServer) popDeploymentID() string {
+ newMinID := atomic.AddInt64(m.minDeploymentID, 1)
+ return strconv.FormatInt(newMinID-1, 10)
+}
+
+func (m *MockServer) init() {
+ defer GinkgoRecover()
+ RegisterFailHandler(func(message string, callerSkip ...int) {
+ log.Errorf("Expect error: %s", message)
+ panic(message)
+ })
+
+ m.sequenceID = new(int64)
+ m.maxDevID = new(int64)
+ m.changeChannel = make(chan []byte)
+ m.minDeploymentID = new(int64)
+ *m.minDeploymentID = 1
+ m.maxDeploymentID = new(int64)
+
+ go m.developerGenerator()
+ go m.developerUpdater()
+ go m.deploymentReplacer()
+
+ // 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.Scope
+ changeSelector := m.params.Scope
+ company := tableRowMap{
+ "kms.company": m.newRow(map[string]string{
+ "id": companyID,
+ "status": "Active",
+ "tenant_id": tenantID,
+ "name": companyID,
+ "display_name": companyID,
+ "_change_selector": changeSelector,
+ }),
+ }
+ snapshotTableRows = append(snapshotTableRows, company)
+
+ // generate snapshot developers
+ for i := 0; i < m.params.NumDevelopers; i++ {
+ developer := m.createDeveloperWithProductAndApp()
+ snapshotTableRows = append(snapshotTableRows, developer)
+ }
+ log.Infof("created %d developers", m.params.NumDevelopers)
+
+ // generate snapshot deployments
+ for i := 0; i < m.params.NumDeployments; i++ {
+ deployment := m.createDeployment()
+ snapshotTableRows = append(snapshotTableRows, deployment)
+ }
+ log.Infof("created %d deployments", m.params.NumDeployments)
+
+ m.snapshotTables[m.params.Scope] = m.concatTableRowMaps(snapshotTableRows...)
+
+ if m.params.NumDevelopers < 10 && m.params.NumDeployments < 10 {
+ log.Debugf("snapshotTables: %v", m.snapshotTables[m.params.Scope])
+ }
+}
+
+// developer, product, application, credential will have the same ID (developerID)
+func (m *MockServer) createDeveloperWithProductAndApp() tableRowMap {
+
+ developerID := m.nextDeveloperID()
+
+ devRows := m.createDeveloper(developerID)
+ productRows := m.createProduct(developerID)
+ appRows := m.createApplication(developerID, developerID, developerID, developerID)
+
+ return m.mergeTableRowMaps(devRows, productRows, appRows)
+}
+
+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")
+ router.HandleFunc("/bundles/{id}", m.sendDeploymentBundle).Methods("GET")
+}
+
+func (m *MockServer) sendDeploymentBundle(w http.ResponseWriter, req *http.Request) {
+ vars := apid.API().Vars(req)
+ w.Write([]byte("/bundles/" + vars["id"]))
+}
+
+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())
+
+ log.Info("sending snapshot")
+ if len(body) < 10000 {
+ log.Debugf("snapshot: %#v", string(body))
+ }
+
+ w.Write(body)
+}
+
+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 since != "" {
+ m.sendChange(w, time.Duration(block)*time.Second)
+ return
+ }
+
+ // todo: the following is just legacy for the existing test in apigeeSync_suite_test
+ developer := m.createDeveloperWithProductAndApp()
+ changeList := m.createInsertChange(developer)
+ body, err := json.Marshal(changeList)
+ if err != nil {
+ log.Errorf("Error generating developer: %v", err)
+ }
+ w.Write(body)
+}
+
+// generate developers w/ product and app
+func (m *MockServer) developerGenerator() {
+
+ for range time.Tick(m.params.AddDeveloperEvery) {
+
+ developer := m.createDeveloperWithProductAndApp()
+ changeList := m.createInsertChange(developer)
+
+ body, err := json.Marshal(changeList)
+ if err != nil {
+ log.Errorf("Error adding developer: %v", err)
+ }
+
+ log.Info("adding developer")
+ log.Debugf("body: %#v", string(body))
+ m.changeChannel <- body
+ }
+}
+
+// update random developers - set username
+func (m *MockServer) developerUpdater() {
+
+ for range time.Tick(m.params.UpdateDeveloperEvery) {
+
+ developerID := m.randomDeveloperID()
+
+ oldDev := m.createDeveloper(developerID)
+ delete(oldDev, "kms.company_developer")
+ newDev := m.createDeveloper(developerID)
+ delete(newDev, "kms.company_developer")
+
+ newRow := newDev["kms.developer"]
+ newRow["username"] = m.stringColumnVal("i_am_not_a_number")
+
+ changeList := m.createUpdateChange(oldDev, newDev)
+
+ body, err := json.Marshal(changeList)
+ if err != nil {
+ log.Errorf("Error updating developer: %v", err)
+ }
+
+ log.Info("updating developer")
+ log.Debugf("body: %#v", string(body))
+ m.changeChannel <- body
+ }
+}
+
+func (m *MockServer) deploymentReplacer() {
+
+ for range time.Tick(m.params.ReplaceDeploymentEvery) {
+
+ // delete
+ oldDep := tableRowMap{}
+ oldDep["edgex.deployment"] = m.newRow(map[string]string{
+ "id": m.popDeploymentID(),
+ })
+ deleteChange := m.createDeleteChange(oldDep)
+
+ // insert
+ newDep := m.createDeployment()
+ insertChange := m.createInsertChange(newDep)
+
+ changeList := m.concatChangeLists(deleteChange, insertChange)
+
+ body, err := json.Marshal(changeList)
+ if err != nil {
+ log.Errorf("Error replacing deployment: %v", err)
+ }
+
+ log.Info("replacing deployment")
+ log.Debugf("body: %#v", string(body))
+ m.changeChannel <- body
+ }
+}
+
+// todo: we could debounce this if necessary
+func (m *MockServer) sendChange(w http.ResponseWriter, timeout time.Duration) {
+ select {
+ case change := <-m.changeChannel:
+ log.Info("sending change to client")
+ w.Write(change)
+ case <-time.After(timeout):
+ log.Info("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))
+ panic(message)
+ })
+}
+
+func (m *MockServer) newRow(keyAndVals map[string]string) (row common.Row) {
+
+ row = common.Row{}
+ for k, v := range keyAndVals {
+ row[k] = m.stringColumnVal(v)
+ }
+
+ // todo: remove this once apidVerifyAPIKey can deal with not having the field
+ row["_change_selector"] = m.stringColumnVal(m.params.Scope)
+
+ return
+}
+
+func (m *MockServer) stringColumnVal(v string) *common.ColumnVal {
+ return &common.ColumnVal{
+ Value: v,
+ Type: 1,
+ }
+}
+
+func (m *MockServer) createDeployment() tableRowMap {
+
+ deploymentID := m.nextDeploymentID()
+ bundleID := generateUUID()
+ port := apid.Config().GetString("api_port")
+
+ urlString := m.params.BundleURI
+ if urlString == "" {
+ urlString = fmt.Sprintf("http://localhost:%s/bundles/%s", port, bundleID)
+ }
+
+ uri, err := url.Parse(urlString)
+ Expect(err).NotTo(HaveOccurred())
+ hashWriter := crc32.NewIEEE()
+ hashWriter.Write([]byte(uri.Path))
+ checkSum := hex.EncodeToString(hashWriter.Sum(nil))
+
+ type bundleConfigJson struct {
+ Name string `json:"name"`
+ URI string `json:"uri"`
+ ChecksumType string `json:"checksumType"`
+ Checksum string `json:"checksum"`
+ }
+
+ bundleJson, err := json.Marshal(bundleConfigJson{
+ Name: uri.Path,
+ URI: urlString,
+ ChecksumType: "crc-32",
+ Checksum: checkSum,
+ })
+ Expect(err).ShouldNot(HaveOccurred())
+
+ rows := tableRowMap{}
+ rows["edgex.deployment"] = m.newRow(map[string]string{
+ "id": deploymentID,
+ "bundle_config_id": bundleID,
+ "apid_cluster_id": m.params.ClusterID,
+ "data_scope_id": m.params.Scope,
+ "bundle_config_json": string(bundleJson),
+ "config_json": "{}",
+ })
+
+ return rows
+}
+
+func (m *MockServer) createDeveloper(developerID string) tableRowMap {
+
+ companyID := m.params.Organization
+ tenantID := m.params.Scope
+
+ rows := tableRowMap{}
+
+ rows["kms.developer"] = m.newRow(map[string]string{
+ "id": developerID,
+ "status": "Active",
+ "tenant_id": tenantID,
+ })
+
+ // map developer onto to existing company
+ rows["kms.company_developer"] = m.newRow(map[string]string{
+ "id": developerID,
+ "tenant_id": tenantID,
+ "company_id": companyID,
+ "developer_id": developerID,
+ })
+
+ return rows
+}
+
+func (m *MockServer) createProduct(productID string) tableRowMap {
+
+ tenantID := m.params.Scope
+
+ environments := fmt.Sprintf("{%s}", m.params.Environment)
+ resources := fmt.Sprintf("{%s}", "/") // todo: what should be here?
+
+ rows := tableRowMap{}
+ rows["kms.api_product"] = m.newRow(map[string]string{
+ "id": productID,
+ "api_resources": resources,
+ "environments": environments,
+ "tenant_id": tenantID,
+ })
+ return rows
+}
+
+func (m *MockServer) createApplication(developerID, productID, applicationID, credentialID string) tableRowMap {
+
+ tenantID := m.params.Scope
+
+ rows := tableRowMap{}
+
+ rows["kms.app"] = m.newRow(map[string]string{
+ "id": applicationID,
+ "developer_id": developerID,
+ "status": "Approved",
+ "tenant_id": tenantID,
+ })
+
+ rows["kms.app_credential"] = m.newRow(map[string]string{
+ "id": credentialID,
+ "app_id": applicationID,
+ "tenant_id": tenantID,
+ "status": "Approved",
+ })
+
+ rows["kms.app_credential_apiproduct_mapper"] = m.newRow(map[string]string{
+ "apiprdt_id": productID,
+ "app_id": applicationID,
+ "appcred_id": credentialID,
+ "status": "Approved",
+ "tenant_id": tenantID,
+ })
+
+ return rows
+}
+
+func (m *MockServer) createInsertChange(newRows tableRowMap) common.ChangeList {
+
+ var changeList = common.ChangeList{}
+ changeList.FirstSequence = m.lastSequenceID()
+ changeList.LastSequence = m.nextSequenceID()
+ 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{}
+ changeList.FirstSequence = m.lastSequenceID()
+ changeList.LastSequence = m.nextSequenceID()
+ 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{}
+ changeList.FirstSequence = m.lastSequenceID()
+ changeList.LastSequence = m.nextSequenceID()
+ 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("overwrite. 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
+}
+
+// create []common.Table from array of tableRowMaps
+func (m *MockServer) concatChangeLists(changeLists ...common.ChangeList) common.ChangeList {
+ result := common.ChangeList{}
+ for _, cl := range changeLists {
+ for _, c := range cl.Changes {
+ result.Changes = append(result.Changes, c)
+ }
+ }
+ return result
+}
