add test, fix bugs
diff --git a/api_test.go b/api_test.go
index 8899f69..bbdd9c4 100644
--- a/api_test.go
+++ b/api_test.go
@@ -311,7 +311,6 @@
CreatedBy: "haoming@google.com",
Updated: time.Now().Format(time.RFC3339),
UpdatedBy: "haoming@google.com",
- BlobFSLocation: "BlobFSLocation",
}
return dep
}
diff --git a/bundle.go b/bundle.go
index b0dc434..660bd53 100644
--- a/bundle.go
+++ b/bundle.go
@@ -128,7 +128,7 @@
for _, dep := range deletedDeployments {
bundleFile := getBlobFilePath(dep.BlobID)
log.Debugf("removing old bundle: %v", bundleFile)
- // TODO Remove from the Database table edgex_blob_available
+ // TODO Remove from the Database table apid_blob_available
safeDelete(bundleFile)
}
}()
diff --git a/data.go b/data.go
index 80beedd..2aa7f13 100644
--- a/data.go
+++ b/data.go
@@ -38,7 +38,6 @@
CreatedBy string
Updated string
UpdatedBy string
- BlobFSLocation string
}
type SQLExec interface {
@@ -78,7 +77,7 @@
func (dbc *dbManager) initDb() error {
_, err := dbc.getDb().Exec(`
- CREATE TABLE IF NOT EXISTS edgex_blob_available (
+ CREATE TABLE IF NOT EXISTS apid_blob_available (
id text primary key,
local_fs_location text NOT NULL
);
@@ -87,7 +86,7 @@
return err
}
- log.Debug("Database table edgex_blob_available created.")
+ log.Debug("Database table apid_blob_available created.")
return nil
}
@@ -99,7 +98,7 @@
rows, err := dbc.getDb().Query(`
SELECT a.bean_blob_id
FROM metadata_runtime_entity_metadata as a
- LEFT JOIN edgex_blob_available as b
+ LEFT JOIN apid_blob_available as b
ON a.bean_blob_id = b.id
WHERE b.id IS NULL;
`)
@@ -118,7 +117,7 @@
rows, err = dbc.getDb().Query(`
SELECT a.resource_blob_id
FROM metadata_runtime_entity_metadata as a
- LEFT JOIN edgex_blob_available as b
+ LEFT JOIN apid_blob_available as b
ON a.resource_blob_id = b.id
WHERE (b.id IS NULL AND a.resource_blob_id IS NOT NULL AND a.resource_blob_id != '');
`)
@@ -153,10 +152,9 @@
a.created_at,
a.created_by,
a.updated_at,
- a.updated_by,
- b.local_fs_location
+ a.updated_by
FROM metadata_runtime_entity_metadata as a
- INNER JOIN edgex_blob_available as b
+ INNER JOIN apid_blob_available as b
ON (a.bean_blob_id = b.id OR a.resource_blob_id = b.id)
;
`)
@@ -185,7 +183,7 @@
func (dbc *dbManager) updateLocalFsLocation(blobId, localFsLocation string) error {
stmt, err := dbc.getDb().Prepare(`
- INSERT OR IGNORE INTO edgex_blob_available (
+ INSERT OR IGNORE INTO apid_blob_available (
id,
local_fs_location
) VALUES (?, ?);`)
@@ -197,11 +195,11 @@
_, err = stmt.Exec(blobId, localFsLocation)
if err != nil {
- log.Errorf("UPDATE edgex_blob_available id {%s} local_fs_location {%s} failed", localFsLocation, err)
+ log.Errorf("UPDATE apid_blob_available id {%s} local_fs_location {%s} failed", localFsLocation, err)
return err
}
- log.Debugf("INSERT edgex_blob_available {%s} local_fs_location {%s} succeeded", blobId, localFsLocation)
+ log.Debugf("INSERT apid_blob_available {%s} local_fs_location {%s} succeeded", blobId, localFsLocation)
return nil
}
@@ -209,7 +207,7 @@
func (dbc *dbManager) getLocalFSLocation(blobId string) (localFsLocation string, err error) {
log.Debugf("Getting the blob file for blobId {%s}", blobId)
- rows, err := dbc.getDb().Query("SELECT local_fs_location FROM edgex_blob_available WHERE id = '" + blobId + "'")
+ rows, err := dbc.getDb().Query("SELECT local_fs_location FROM apid_blob_available WHERE id = '" + blobId + "'")
if err != nil {
log.Errorf("SELECT local_fs_location failed %v", err)
return "", err
@@ -244,7 +242,6 @@
&dep.CreatedBy,
&dep.Updated,
&dep.UpdatedBy,
- &dep.BlobFSLocation,
)
if err != nil {
return nil, err
diff --git a/data_test.go b/data_test.go
index 208dac8..29f777a 100644
--- a/data_test.go
+++ b/data_test.go
@@ -64,9 +64,9 @@
})
It("should succefully initialized tables", func() {
- // edgex_blob_available
+ // apid_blob_available
rows, err := testDbMan.getDb().Query(`
- SELECT count(*) from edgex_blob_available;
+ SELECT count(*) from apid_blob_available;
`)
Expect(err).Should(Succeed())
defer rows.Close()
@@ -92,9 +92,9 @@
err := testDbMan.updateLocalFsLocation(readyBlobId, readyblobLocalFs)
Expect(err).Should(Succeed())
- // edgex_blob_available
+ // apid_blob_available
rows, err := testDbMan.getDb().Query(`
- SELECT count(*) from edgex_blob_available;
+ SELECT count(*) from apid_blob_available;
`)
Expect(err).Should(Succeed())
defer rows.Close()
@@ -110,7 +110,7 @@
err := testDbMan.updateLocalFsLocation(readyBlobId, readyblobLocalFs)
Expect(err).Should(Succeed())
- // edgex_blob_available
+ // apid_blob_available
location, err := testDbMan.getLocalFSLocation(readyBlobId)
Expect(err).Should(Succeed())
Expect(location).Should(Equal(readyblobLocalFs))
@@ -124,7 +124,6 @@
Expect(err).Should(Succeed())
Expect(len(deps)).Should(Equal(1))
Expect(deps[0].BlobID).Should(Equal(readyBlobId))
- Expect(deps[0].BlobFSLocation).Should(Equal(readyblobLocalFs))
})
It("should succefully get unready blob ids", func() {
diff --git a/listener.go b/listener.go
index c2b05df..701d3a1 100644
--- a/listener.go
+++ b/listener.go
@@ -22,7 +22,7 @@
const (
APIGEE_SYNC_EVENT = "ApigeeSync"
- CONFIG_METADATA_TABLE = "project.runtime_blob_metadata"
+ CONFIG_METADATA_TABLE = "metadata.runtime_entity_metadata"
)
func (h *apigeeSyncHandler) initListener(services apid.Services) {
@@ -80,7 +80,7 @@
func (h *apigeeSyncHandler) startupOnExistingDatabase() {
// start bundle downloads that didn't finish
go func() {
- // create edgex_blob_available table
+ // create apid_blob_available table
h.dbMan.initDb()
blobIds, err := h.dbMan.getUnreadyBlobs()
@@ -127,8 +127,8 @@
}
*/
- for _, dep := range insertedDeployments {
- go h.bundleMan.queueDownloadRequest(&dep)
+ for i := range insertedDeployments {
+ go h.bundleMan.queueDownloadRequest(&insertedDeployments[i])
}
// clean up old bundles
diff --git a/listener_test.go b/listener_test.go
index 0cced6f..ad0b258 100644
--- a/listener_test.go
+++ b/listener_test.go
@@ -21,6 +21,7 @@
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"math/rand"
+ "reflect"
"time"
)
@@ -37,6 +38,7 @@
dummyDbMan = &dummyDbManager{}
dummyBundleMan = &dummyBundleManager{
requestChan: make(chan *DownloadRequest),
+ depChan: make(chan *DataDeployment),
}
testHandler = &apigeeSyncHandler{
dbMan: dummyDbMan,
@@ -56,7 +58,7 @@
// init unready blob ids
unreadyBlobIds := make([]string, 0)
blobMap := make(map[string]int)
- for i := 0; i < rand.Intn(10); i++ {
+ for i := 0; i < 1+rand.Intn(10); i++ {
id := GenerateUUID()
blobMap[id] = 1
unreadyBlobIds = append(unreadyBlobIds, id)
@@ -99,10 +101,41 @@
})
+ Context("Change list", func() {
+
+ It("Insert event shoud enqueue download requests for all inserted deployments", func() {
+ // emit change event
+ changes := make([]common.Change, 0)
+ deployments := make(map[string]DataDeployment)
+ for i := 0; i < 1+rand.Intn(10); i++ {
+ dep := makeTestDeployment()
+ change := common.Change{
+ Operation: common.Insert,
+ Table: CONFIG_METADATA_TABLE,
+ NewRow: rowFromDeployment(dep),
+ }
+ changes = append(changes, change)
+ deployments[dep.ID] = *dep
+ }
+
+ changeList := &common.ChangeList{
+ Changes: changes,
+ }
+
+ apid.Events().Emit(APIGEE_SYNC_EVENT, changeList)
+
+ for i := 0; i < len(changes); i++ {
+ dep := <-dummyBundleMan.depChan
+ Expect(reflect.DeepEqual(deployments[dep.ID], *dep)).Should(BeTrue())
+ delete(deployments, dep.ID)
+ }
+ })
+ })
})
type dummyBundleManager struct {
requestChan chan *DownloadRequest
+ depChan chan *DataDeployment
}
func (bm *dummyBundleManager) initializeBundleDownloading() {
@@ -110,7 +143,7 @@
}
func (bm *dummyBundleManager) queueDownloadRequest(dep *DataDeployment) {
-
+ bm.depChan <- dep
}
func (bm *dummyBundleManager) enqueueRequest(req *DownloadRequest) {
@@ -126,3 +159,22 @@
func (bm *dummyBundleManager) Close() {
}
+
+func rowFromDeployment(dep *DataDeployment) common.Row {
+ row := common.Row{}
+ row["id"] = &common.ColumnVal{Value: dep.ID}
+ row["organization_id"] = &common.ColumnVal{Value: dep.OrgID}
+ row["environment_id"] = &common.ColumnVal{Value: dep.EnvID}
+ row["bean_blob_id"] = &common.ColumnVal{Value: dep.BlobID}
+ row["resource_blob_id"] = &common.ColumnVal{Value: dep.BlobResourceID}
+ row["type"] = &common.ColumnVal{Value: dep.Type}
+ row["name"] = &common.ColumnVal{Value: dep.Name}
+ row["revision"] = &common.ColumnVal{Value: dep.Revision}
+ row["path"] = &common.ColumnVal{Value: dep.Path}
+ row["created_at"] = &common.ColumnVal{Value: dep.Created}
+ row["created_by"] = &common.ColumnVal{Value: dep.CreatedBy}
+ row["updated_at"] = &common.ColumnVal{Value: dep.Updated}
+ row["updated_by"] = &common.ColumnVal{Value: dep.UpdatedBy}
+
+ return row
+}
