func messagePartitionWriter(a *assert.Assertions, store *messagePartition, n int, done chan bool) {
	for i := 1; i <= n; i++ {
		msg := []byte("Hello " + strconv.Itoa(i))
		a.NoError(store.Store(uint64(i), msg))
	}
	done <- true
}

// Assert Order type
func assertOrder(assert *assert.Assertions, order *OrderReply) {
	assert.EqualValues(123, order.OrderId)
	assert.Equal("test", order.ResultCode)
	assert.Equal("test", order.OrderState)
	assert.Equal("test", order.ActionState)
	assert.Equal("test", order.Message)
}

// overrideFromLayer performs the sequential override and low-level checks.
//
// First assignment is made on layer l for path firstPath with value firstValue,
// the second one on the override layer (i.e., with the Set() function)
// for path secondPath with value secondValue.
//
// firstPath and secondPath can include an arbitrary number of dots to indicate
// a nested element.
//
// After each assignment, the value is checked, retrieved both by its full path
// and by its key sequence (successive maps).
func overrideFromLayer(l layer, assert *assert.Assertions, firstPath string, firstValue interface{}, secondPath string, secondValue interface{}) *Viper {
	v := New()
	firstKeys := strings.Split(firstPath, v.keyDelim)
	if assert == nil ||
		len(firstKeys) == 0 || len(firstKeys[0]) == 0 {
		return v
	}

	// Set and check first value
	switch l {
	case defaultLayer:
		v.SetDefault(firstPath, firstValue)
	case overrideLayer:
		v.Set(firstPath, firstValue)
	default:
		return v
	}
	assert.Equal(firstValue, v.Get(firstPath))
	deepCheckValue(assert, v, l, firstKeys, firstValue)

	// Override and check new value
	secondKeys := strings.Split(secondPath, v.keyDelim)
	if len(secondKeys) == 0 || len(secondKeys[0]) == 0 {
		return v
	}
	v.Set(secondPath, secondValue)
	assert.Equal(secondValue, v.Get(secondPath))
	deepCheckValue(assert, v, overrideLayer, secondKeys, secondValue)

	return v
}

func assertChannelContainsMessage(a *assert.Assertions, c <-chan *protocol.Message, msg []byte) {
	select {
	case m := <-c:
		a.Equal(string(msg), string(m.Body))
	case <-time.After(time.Millisecond * 5):
		a.Fail("No message received")
	}
}

// ExpectDone waits to receive a value in the doneChannel for at least a second
// or fails the test.
func ExpectDone(a *assert.Assertions, doneChannel chan bool) {
	select {
	case <-doneChannel:
		return
	case <-time.After(time.Second):
		a.Fail("timeout in expectDone")
	}
}

func writeJSONToService(service annotations.Service, pathToJSONFile string, contentUUID string, assert *assert.Assertions) {
	f, err := os.Open(pathToJSONFile)
	assert.NoError(err)
	dec := json.NewDecoder(f)
	annotation, errr := service.DecodeJSON(dec)
	assert.NoError(errr)
	errrr := service.Write(contentUUID, annotation)
	assert.NoError(errrr)
}

func writeJSONToService(service baseftrwapp.Service, pathToJSONFile string, assert *assert.Assertions) {
	f, err := os.Open(pathToJSONFile)
	assert.NoError(err)
	dec := json.NewDecoder(f)
	inst, _, errr := service.DecodeJSON(dec)
	assert.NoError(errr)
	errrr := service.Write(inst)
	assert.NoError(errrr)
}

func assertChannelContainsMessage(a *assert.Assertions, c chan MsgAndRoute, msg []byte) {
	//log.Println("DEBUG: start assertChannelContainsMessage-> select")
	select {
	case msgBack := <-c:
		a.Equal(string(msg), string(msgBack.Message.Body))
	case <-time.After(time.Millisecond):
		a.Fail("No message received")
	}
}

func messagePartitionReader(name string, a *assert.Assertions, store *MessagePartition, n int, done chan bool) {
	lastReadMessage := 0
	for lastReadMessage < n {

		msgC := make(chan MessageAndId)
		errorC := make(chan error)

		log.WithFields(log.Fields{
			"module":      "testing",
			"name":        name,
			"lastReadMsg": lastReadMessage + 1,
		}).Debug("Start fetching at")

		store.Fetch(FetchRequest{
			Partition:     "myMessages",
			StartId:       uint64(lastReadMessage + 1),
			Direction:     1,
			Count:         math.MaxInt32,
			MessageC:      msgC,
			ErrorCallback: errorC,
			StartCallback: make(chan int, 1),
		})

	fetch:

		for {
			select {
			case msgAndId, open := <-msgC:
				if !open {

					log.WithFields(log.Fields{
						"module":      "testing",
						"name":        name,
						"lastReadMsg": lastReadMessage,
					}).Debug("Stop fetching at")

					break fetch
				}
				a.Equal(lastReadMessage+1, int(msgAndId.Id))
				lastReadMessage = int(msgAndId.Id)
			case err := <-errorC:
				a.Fail("received error", err.Error())
				<-done
				return
			}
		}

	}
	log.WithFields(log.Fields{
		"module":      "testing",
		"name":        name,
		"lastReadMsg": lastReadMessage,
	}).Debug("Ready got id")

	done <- true
}

func expectMessages(a *assert.Assertions, msgChannel chan []byte, message ...string) {
	for _, m := range message {
		select {
		case msg := <-msgChannel:
			a.Equal(m, string(msg))
		case <-time.After(time.Millisecond * 100):
			a.Fail("timeout: " + m)
		}
	}
}

func verifyLocalFileFromDefaultAsset(dir string, assert *assert.Assertions) {
	assert.Nil(verifyLocalFile(dir, "index.html"))
	assert.Nil(verifyLocalFile(dir, "site.js"))
	assert.Nil(verifyLocalFile(dir, "assets/1.txt"))
	assert.Nil(verifyLocalFile(dir, "assets/2.txt"))
	assert.Nil(verifyLocalFile(dir, "assets/more/3.txt"))
	assert.Nil(verifyLocalFile(dir, "assets/four/4.txt"))
	assert.Nil(verifyLocalFile(dir, "assets/fivesix/5.txt"))
	assert.Nil(verifyLocalFile(dir, "assets/fivesix/6.txt"))
}

func loadJson(filename string, a *assert.Assertions) map[string]string {
	var data map[string]string
	content, err := ioutil.ReadFile("./test.json")

	if err != nil {
		a.Fail("Failed to read file ", err)
	}

	json.Unmarshal(content, &data)
	return data
}

func runComplianceTest(assert *assert.Assertions, filename string) {
	var testSuites []TestSuite
	data, err := ioutil.ReadFile(filename)
	if assert.Nil(err) {
		err := json.Unmarshal(data, &testSuites)
		if assert.Nil(err) {
			for _, testsuite := range testSuites {
				runTestSuite(assert, testsuite, filename)
			}
		}
	}
}

func messagePartitionReader(name string, a *assert.Assertions, mStore *messagePartition, n int, done chan bool) {
	lastReadMessage := 0

	for lastReadMessage < n {
		msgC := make(chan *store.FetchedMessage, 10)
		errorC := make(chan error)

		log.WithFields(log.Fields{
			"module":      "testing",
			"name":        name,
			"lastReadMsg": lastReadMessage + 1,
		}).Debug("Start fetching")

		mStore.Fetch(&store.FetchRequest{
			Partition: "myMessages",
			StartID:   uint64(lastReadMessage + 1),
			Direction: 1,
			Count:     math.MaxInt32,
			MessageC:  msgC,
			ErrorC:    errorC,
			StartC:    make(chan int, 1),
		})

	FETCH:
		for {
			select {
			case msgAndID, open := <-msgC:
				if !open {
					log.WithFields(log.Fields{
						"module":      "testing",
						"name":        name,
						"lastReadMsg": lastReadMessage,
					}).Debug("Stop fetching")
					break FETCH
				}
				a.Equal(lastReadMessage+1, int(msgAndID.ID), "Reader: "+name)
				lastReadMessage = int(msgAndID.ID)
			case err := <-errorC:
				a.Fail("received error", err.Error())
				<-done
				return
			}
		}
	}

	log.WithFields(log.Fields{
		"module":      "testing",
		"name":        name,
		"lastReadMsg": lastReadMessage,
	}).Debug("Ready got id")

	done <- true
}

func getServices(t *testing.T, assert *assert.Assertions, db neoutils.NeoConnection) (baseftrwapp.Service, baseftrwapp.Service, baseftrwapp.Service, baseftrwapp.Service, baseftrwapp.Service) {
	peopleRW := people.NewCypherPeopleService(db)
	assert.NoError(peopleRW.Initialise())
	organisationRW := organisations.NewCypherOrganisationService(db)
	assert.NoError(organisationRW.Initialise())
	membershipsRW := memberships.NewCypherMembershipService(db)
	assert.NoError(membershipsRW.Initialise())
	rolesRW := roles.NewCypherDriver(db)
	assert.NoError(rolesRW.Initialise())
	financialInstrumentsRW := financialinstruments.NewCypherFinancialInstrumentService(db)
	assert.NoError(financialInstrumentsRW.Initialise())
	return peopleRW, organisationRW, membershipsRW, rolesRW, financialInstrumentsRW
}

func newTestableProvisioner(assert *assert.Assertions, options volume.VolumeOptions) volume.Provisioner {
	tmpDir, err := utiltesting.MkTmpdir("flockervolumeTest")
	assert.NoError(err, fmt.Sprintf("can't make a temp dir: %v", err))

	plugMgr := volume.VolumePluginMgr{}
	plugMgr.InitPlugins(ProbeVolumePlugins(), volumetest.NewFakeVolumeHost(tmpDir, nil, nil))

	plug, err := plugMgr.FindPluginByName(pluginName)
	assert.NoError(err, "Can't find the plugin by name")

	provisioner, err := plug.(*flockerPlugin).newProvisionerInternal(options, &fakeFlockerUtil{})

	return provisioner
}

func mockGetStateNodes(assert *assert.Assertions, data []byte) *Client {

	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		assert.Equal("GET", r.Method)
		assert.Equal("/v1/state/nodes", r.URL.Path)
		w.Write(data)
	}))

	host, port, err := getHostAndPortFromTestServer(ts)
	assert.NoError(err)

	c := newFlockerTestClient(host, port)

	return c
}

func writeClassifedByRelationship(db neoutils.NeoConnection, contentId string, conceptId string, lifecycle string, t *testing.T, assert *assert.Assertions) {
	var annotateQuery string
	var qs []*neoism.CypherQuery

	if lifecycle == "" {
		annotateQuery = `
                MERGE (content:Thing{uuid:{contentId}})
                MERGE (upp:Identifier:UPPIdentifier{value:{conceptId}})
                MERGE (upp)-[:IDENTIFIES]->(concept:Thing) ON CREATE SET concept.uuid = {conceptId}
                MERGE (content)-[pred:IS_CLASSIFIED_BY {platformVersion:'v1'}]->(concept)
          `
		qs = []*neoism.CypherQuery{
			{
				Statement:  annotateQuery,
				Parameters: neoism.Props{"contentId": contentId, "conceptId": conceptId},
			},
		}
	} else {
		annotateQuery = `
                MERGE (content:Thing{uuid:{contentId}})
                MERGE (upp:Identifier:UPPIdentifier{value:{conceptId}})
                MERGE (upp)-[:IDENTIFIES]->(concept:Thing) ON CREATE SET concept.uuid = {conceptId}
                MERGE (content)-[pred:IS_CLASSIFIED_BY {platformVersion:'v1', lifecycle: {lifecycle}}]->(concept)
          `
		qs = []*neoism.CypherQuery{
			{
				Statement:  annotateQuery,
				Parameters: neoism.Props{"contentId": contentId, "conceptId": conceptId, "lifecycle": lifecycle},
			},
		}

	}
	err := db.CypherBatch(qs)
	assert.NoError(err)
}

func deleteAllViaService(db neoutils.CypherRunner, assert *assert.Assertions, annotationsRW annotations.Service) {
	_, err := annotationsRW.Delete(contentUUID)
	assert.Nil(err)

	qs := []*neoism.CypherQuery{
		{
			Statement: fmt.Sprintf("MATCH (c:Thing {uuid: '%v'})-[rel]-(o) DELETE c, rel ", contentUUID),
		},
		{
			Statement: fmt.Sprintf("MATCH (c:Thing {uuid: '%v'}) DELETE c ", contentUUID),
		},
	}

	err = db.CypherBatch(qs)
	assert.NoError(err)
}

func checkRelationship(assert *assert.Assertions, contentID string, platformVersion string) {
	countQuery := `Match (t:Thing {uuid: {contentID}})-[r {lifecycle: {lifecycle}}]-(x) return count(r) as c`

	results := []struct {
		Count int `json:"c"`
	}{}

	qs := &neoism.CypherQuery{
		Statement:  countQuery,
		Parameters: neoism.Props{"contentID": contentID, "lifecycle": "annotations-" + platformVersion},
		Result:     &results,
	}

	err := annotationsDriver.conn.CypherBatch([]*neoism.CypherQuery{qs})
	assert.NoError(err)
	assert.Equal(1, len(results), "More results found than expected!")
	assert.Equal(1, results[0].Count, "No Relationship with Lifecycle found!")
}