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