// Make sure new labelgraph data have different IDs.
func TestNewLabelgraphDifferent(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Add data
	config := dvid.NewConfig()
	dataservice1, err := datastore.NewData(uuid, dtype, "lg1", config)
	if err != nil {
		t.Errorf("Error creating new labelgraph instance 1: %v\n", err)
	}
	data1, ok := dataservice1.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 1 is not labelgraph.Data\n")
	}
	dataservice2, err := datastore.NewData(uuid, dtype, "lg2", config)
	if err != nil {
		t.Errorf("Error creating new labelgraph instance 2: %v\n", err)
	}
	data2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not labelgraph.Data\n")
	}
	if data1.InstanceID() == data2.InstanceID() {
		t.Errorf("Instance IDs should be different: %d == %d\n",
			data1.InstanceID(), data2.InstanceID())
	}
}

// Make sure new keyvalue data have different IDs.
func TestNewKeyvalueDifferent(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Add data
	config := dvid.NewConfig()
	dataservice1, err := datastore.NewData(uuid, kvtype, "instance1", config)
	if err != nil {
		t.Errorf("Error creating new keyvalue instance: %v\n", err)
	}
	kv1, ok := dataservice1.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 1 is not keyvalue.Data\n")
	}
	if kv1.DataName() != "instance1" {
		t.Errorf("New keyvalue data instance name set incorrectly: %q != %q\n",
			kv1.DataName(), "instance1")
	}

	dataservice2, err := datastore.NewData(uuid, kvtype, "instance2", config)
	if err != nil {
		t.Errorf("Error creating new keyvalue instance: %v\n", err)
	}
	kv2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not keyvalue.Data\n")
	}

	if kv1.InstanceID() == kv2.InstanceID() {
		t.Errorf("Instance IDs should be different: %d == %d\n",
			kv1.InstanceID(), kv2.InstanceID())
	}
}

func TestROICreateAndSerialize(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Add data
	config := dvid.NewConfig()
	dataservice1, err := datastore.NewData(uuid, roitype, "myroi", config)
	if err != nil {
		t.Errorf("Error creating new roi instance: %v\n", err)
	}
	roi1, ok := dataservice1.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 1 is not roi.Data\n")
	}
	if roi1.DataName() != "myroi" {
		t.Errorf("New roi data instance name set incorrectly: %q != %q\n",
			roi1.DataName(), "myroi")
	}

	config.Set("BlockSize", "15,16,17")
	dataservice2, err := datastore.NewData(uuid, roitype, "myroi2", config)
	if err != nil {
		t.Errorf("Error creating new roi instance: %v\n", err)
	}
	roi2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not roi.Data\n")
	}

	if roi1.InstanceID() == roi2.InstanceID() {
		t.Errorf("Instance IDs should be different: %d == %d\n",
			roi1.InstanceID(), roi2.InstanceID())
	}

	// Test persistence of storage.
	roi2.MinZ = 13
	roi2.MaxZ = 3098
	gobBytes, err := roi2.GobEncode()
	if err != nil {
		t.Fatalf("Could not Gob encode roi: %v\n", err)
	}

	var received Data
	if err = received.GobDecode(gobBytes); err != nil {
		t.Fatalf("Could not decode Gob-encoded roi: %v\n", err)
	}

	if !roi2.Data.Equals(received.Data) {
		t.Errorf("ROI base Data has bad roundtrip:\nOriginal:\n%v\nReceived:\n%v\n",
			*(roi2.Data), *(received.Data))
	}

	if !reflect.DeepEqual(roi2.Properties, received.Properties) {
		t.Errorf("ROI extended properties has bad roundtrip:\nOriginal:\n%v\nReceived:\n%v\n",
			roi2.Properties, received.Properties)
	}
}

func repoNewDataHandler(c web.C, w http.ResponseWriter, r *http.Request) {
	uuid := c.Env["uuid"].(dvid.UUID)
	config := dvid.NewConfig()
	if err := config.SetByJSON(r.Body); err != nil {
		BadRequest(w, r, fmt.Sprintf("Error decoding POSTed JSON config for 'new': %v", err))
		return
	}

	// Make sure that the passed configuration has data type and instance name.
	typename, found, err := config.GetString("typename")
	if !found || err != nil {
		BadRequest(w, r, "POST on repo endpoint requires specification of valid 'typename'")
		return
	}
	dataname, found, err := config.GetString("dataname")
	if !found || err != nil {
		BadRequest(w, r, "POST on repo endpoint requires specification of valid 'dataname'")
		return
	}
	typeservice, err := datastore.TypeServiceByName(dvid.TypeString(typename))
	if err != nil {
		BadRequest(w, r, err)
		return
	}
	_, err = datastore.NewData(uuid, typeservice, dvid.InstanceName(dataname), config)
	if err != nil {
		BadRequest(w, r, err)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	fmt.Fprintf(w, "{%q: 'Added %s [%s] to node %s'}", "result", dataname, typename, uuid)
}

func TestRequests(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, syntype, "mysynapses", config)
	if err != nil {
		t.Fatalf("Error creating new data instance: %v\n", err)
	}
	data, ok := dataservice.(*Data)
	if !ok {
		t.Fatalf("Returned new data instance is not synapse.Data\n")
	}

	// PUT first batch of synapses
	testJSON, err := json.Marshal(testData)
	if err != nil {
		t.Fatal(err)
	}
	url1 := fmt.Sprintf("%snode/%s/%s/elements", server.WebAPIPath, uuid, data.DataName())
	server.TestHTTP(t, "POST", url1, strings.NewReader(string(testJSON)))

	// GET synapses back within superset bounding box and make sure all data is there.
	testResponse(t, testData, "%snode/%s/%s/elements/1000_1000_1000/0_0_0", server.WebAPIPath, uuid, data.DataName())

	// Test subset GET
	testResponse(t, expected3, "%snode/%s/%s/elements/5_5_5/126_60_97", server.WebAPIPath, uuid, data.DataName())

	// Test Tag 1
	tag := Tag("Synapse2")
	synapse2 := getTag(tag, testData)
	testResponse(t, synapse2, "%snode/%s/%s/tag/%s?relationships=true", server.WebAPIPath, uuid, data.DataName(), tag)

	// Test Tag 2
	tag2 := Tag("Zlt90")
	zlt90 := getTag(tag2, testData)
	testResponse(t, zlt90, "%snode/%s/%s/tag/%s?relationships=true", server.WebAPIPath, uuid, data.DataName(), tag2)

	// Test move
	url5 := fmt.Sprintf("%snode/%s/%s/move/127_63_99/127_64_100", server.WebAPIPath, uuid, data.DataName())
	server.TestHTTP(t, "POST", url5, nil)
	testResponse(t, afterMove, "%snode/%s/%s/elements/1000_1000_1000/0_0_0", server.WebAPIPath, uuid, data.DataName())

	// --- check tag
	synapse2 = getTag(tag, afterMove)
	testResponse(t, synapse2, "%snode/%s/%s/tag/%s?relationships=true", server.WebAPIPath, uuid, data.DataName(), tag)

	// Test delete
	url6 := fmt.Sprintf("%snode/%s/%s/element/127_64_100", server.WebAPIPath, uuid, data.DataName())
	server.TestHTTP(t, "DELETE", url6, nil)
	testResponse(t, afterDelete, "%snode/%s/%s/elements/1000_1000_1000/0_0_0", server.WebAPIPath, uuid, data.DataName())

	// --- check tag
	synapse2 = getTag(tag, afterDelete)
	testResponse(t, synapse2, "%snode/%s/%s/tag/%s?relationships=true", server.WebAPIPath, uuid, data.DataName(), tag)
}

// Creates a new data instance for labelblk
func newDataInstance(uuid dvid.UUID, t *testing.T, name dvid.InstanceName) *Data {
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, labelsT, name, config)
	if err != nil {
		t.Errorf("Unable to create labelblk instance %q: %v\n", name, err)
	}
	labels, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast labels data service into Data\n")
	}
	return labels
}

func makeGrayscale(uuid dvid.UUID, t *testing.T, name dvid.InstanceName) *imageblk.Data {
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, grayscaleT, name, config)
	if err != nil {
		t.Errorf("Unable to create grayscale instance %q: %v\n", name, err)
	}
	grayscale, ok := dataservice.(*imageblk.Data)
	if !ok {
		t.Errorf("Can't cast data service into imageblk Data\n")
	}
	return grayscale
}

func TestBasic(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	config := dvid.NewConfig()
	_, err := datastore.NewData(uuid, dtype, "instance1", config)
	if err != nil {
		t.Errorf("Error creating new multichan16 instance: %v\n", err)
	}
}

func TestROIPartition(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	// Create the ROI dataservice.
	uuid, versionID := initTestRepo()

	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, roitype, "roi", config)
	if err != nil {
		t.Errorf("Error creating new roi instance: %v\n", err)
	}
	data, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Returned new data instance is not roi.Data\n")
	}

	// PUT an ROI
	roiRequest := fmt.Sprintf("%snode/%s/%s/roi", server.WebAPIPath, uuid, data.DataName())
	req, err := http.NewRequest("POST", roiRequest, getSpansJSON(testSpans))
	if err != nil {
		t.Errorf("Unsuccessful POST request (%s): %v\n", roiRequest, err)
	}
	ctx := datastore.NewVersionedCtx(data, versionID)
	w := httptest.NewRecorder()
	data.ServeHTTP(uuid, ctx, w, req)
	if w.Code != http.StatusOK {
		t.Errorf("Bad server response roi POST, status %s, for roi %q\n", w.Code, data.DataName())
	}

	// Request the standard subvolume partitioning
	partitionReq := fmt.Sprintf("%snode/%s/%s/partition?batchsize=5&optimized=true", server.WebAPIPath, uuid,
		data.DataName())
	req, err = http.NewRequest("GET", partitionReq, nil)
	if err != nil {
		t.Errorf("Unsuccessful GET request (%s): %v\n", partitionReq, err)
	}
	w = httptest.NewRecorder()
	data.ServeHTTP(uuid, ctx, w, req)
	if w.Code != http.StatusOK {
		t.Errorf("Bad server response roi GET, status %s, for roi %q\n", w.Code, data.DataName())
	}
	var subvolJSON, expectedJSON interface{}
	response := w.Body.Bytes()
	if err := json.Unmarshal(response, &subvolJSON); err != nil {
		t.Errorf("Can't unmarshal JSON: %s\n", w.Body.Bytes())
	}
	json.Unmarshal([]byte(expectedPartition), &expectedJSON)
	if !reflect.DeepEqual(subvolJSON, expectedJSON) {
		t.Errorf("Error doing optimized subvolume partitioning.  Got bad result:\n%s\n",
			string(response))
	}
}

// Test added after error in getting two paths to the same ancestor k/v after merge.
func TestDiamondGetOnMerge(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, kvtype, "mergetest", config)
	if err != nil {
		t.Fatalf("Error creating new keyvalue instance: %v\n", err)
	}
	data, ok := dataservice.(*Data)
	if !ok {
		t.Fatalf("Returned new data instance is not roi.Data\n")
	}

	// PUT a value
	key1 := "mykey"
	value1 := "some stuff"
	key1req := fmt.Sprintf("%snode/%s/%s/key/%s", server.WebAPIPath, uuid, data.DataName(), key1)
	server.TestHTTP(t, "POST", key1req, strings.NewReader(value1))

	if err = datastore.Commit(uuid, "my commit msg", []string{"stuff one", "stuff two"}); err != nil {
		t.Errorf("Unable to lock root node %s: %v\n", uuid, err)
	}
	uuid2, err := datastore.NewVersion(uuid, "first child", nil)
	if err != nil {
		t.Fatalf("Unable to create 1st child off root %s: %v\n", uuid, err)
	}
	if err = datastore.Commit(uuid2, "first child", nil); err != nil {
		t.Errorf("Unable to commit node %s: %v\n", uuid2, err)
	}
	uuid3, err := datastore.NewVersion(uuid, "second child", nil)
	if err != nil {
		t.Fatalf("Unable to create 2nd child off root %s: %v\n", uuid, err)
	}
	if err = datastore.Commit(uuid3, "second child", nil); err != nil {
		t.Errorf("Unable to commit node %s: %v\n", uuid3, err)
	}

	child, err := datastore.Merge([]dvid.UUID{uuid2, uuid3}, "merging stuff", datastore.MergeConflictFree)
	if err != nil {
		t.Errorf("Error doing merge: %v\n", err)
	}

	// We should be able to see just the original uuid value of the k/v
	childreq := fmt.Sprintf("%snode/%s/%s/key/%s", server.WebAPIPath, child, data.DataName(), key1)
	returnValue := server.TestHTTP(t, "GET", childreq, nil)
	if string(returnValue) != value1 {
		t.Errorf("Error on merged child, key %q: expected %q, got %q\n", key1, value1, string(returnValue))
	}
}

// ForegroundROI creates a new ROI by determining all non-background blocks.
func (d *Data) ForegroundROI(req datastore.Request, reply *datastore.Response) error {
	if d.Values.BytesPerElement() != 1 {
		return fmt.Errorf("Foreground ROI command only implemented for 1 byte/voxel data!")
	}

	// Parse the request
	var uuidStr, dataName, cmdStr, destName, backgroundStr string
	req.CommandArgs(1, &uuidStr, &dataName, &cmdStr, &destName, &backgroundStr)

	// Get the version and repo
	uuid, versionID, err := datastore.MatchingUUID(uuidStr)
	if err != nil {
		return err
	}
	if err = datastore.AddToNodeLog(uuid, []string{req.Command.String()}); err != nil {
		return err
	}

	// Use existing destination data or a new ROI data.
	var dest *roi.Data
	dest, err = roi.GetByUUID(uuid, dvid.InstanceName(destName))
	if err != nil {
		config := dvid.NewConfig()
		typeservice, err := datastore.TypeServiceByName("roi")
		if err != nil {
			return err
		}
		dataservice, err := datastore.NewData(uuid, typeservice, dvid.InstanceName(destName), config)
		if err != nil {
			return err
		}
		var ok bool
		dest, ok = dataservice.(*roi.Data)
		if !ok {
			return fmt.Errorf("Could not create ROI data instance")
		}
	}

	// Asynchronously process the voxels.
	background, err := dvid.StringToPointNd(backgroundStr, ",")
	if err != nil {
		return err
	}
	go d.foregroundROI(versionID, dest, background)

	return nil
}

// check subgraph endpoint
func TestLabelgraphPostAndDelete(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	// Create the ROI dataservice.
	uuid, _ := initTestRepo()

	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, dtype, "lg", config)
	if err != nil {
		t.Errorf("Error creating new labelgraph instance: %v\n", err)
	}
	data, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Returned new data instance is not labelgraph.Data\n")
	}

	// PUT a labelraph
	subgraphRequest := fmt.Sprintf("%snode/%s/%s/subgraph", server.WebAPIPath, uuid, data.DataName())
	server.TestHTTP(t, "POST", subgraphRequest, getGraphJSON())

	// Get back the labelgraph
	returnedData := server.TestHTTP(t, "GET", subgraphRequest, nil)
	retgraph, err := loadGraphJSON(returnedData)
	if err != nil {
		t.Errorf("Error on getting back JSON from roi GET: %v\n", err)
	}

	// Make sure the two are the same.
	if !reflect.DeepEqual(retgraph, getTestGraph()) {
		t.Errorf("Bad PUT/GET ROI roundtrip\nOriginal:\n%s\nReturned:\n%s\n", getTestGraph(), retgraph)
	}

	// Delete the labelgraph
	_ = server.TestHTTP(t, "DELETE", subgraphRequest, nil)

	// Subgraph should now be empty
	returnedData = server.TestHTTP(t, "GET", subgraphRequest, nil)
	expectedResp := "{\"Transactions\":[],\"Vertices\":[],\"Edges\":[]}"
	if string(returnedData) != expectedResp {
		t.Errorf("Bad ROI after ROI delete.  Should be %s got: %s\n", expectedResp, string(returnedData))
	}
}

func TestMultiscale2dRepoPersistence(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	// Make source
	uuid, _ := initTestRepo()
	makeGrayscale(uuid, t, "grayscale")

	// Make labels and set various properties
	config := dvid.NewConfig()
	config.Set("Placeholder", "true")
	config.Set("Format", "jpg")
	config.Set("Source", "grayscale")
	dataservice, err := datastore.NewData(uuid, mstype, "myimagetile", config)
	if err != nil {
		t.Errorf("Unable to create imagetile instance: %v\n", err)
	}
	msdata, ok := dataservice.(*Data)
	if !ok {
		t.Fatalf("Can't cast imagetile data service into imagetile.Data\n")
	}
	oldData := *msdata

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, msdata); err != nil {
		t.Fatalf("Unable to save repo during imagetile persistence test: %v\n", err)
	}
	datastore.CloseReopenTest()

	dataservice2, err := datastore.GetDataByUUID(uuid, "myimagetile")
	if err != nil {
		t.Fatalf("Can't get keyvalue instance from reloaded test db: %v\n", err)
	}
	msdata2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not imagetile.Data\n")
	}

	if !reflect.DeepEqual(oldData.Properties, msdata2.Properties) {
		t.Errorf("Expected properties %v, got %v\n", oldData.Properties, msdata2.Properties)
	}
}

func TestROIPostAndDelete(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	// Create the ROI dataservice.
	uuid, _ := initTestRepo()

	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, roitype, "roi", config)
	if err != nil {
		t.Errorf("Error creating new roi instance: %v\n", err)
	}
	data, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Returned new data instance is not roi.Data\n")
	}

	// PUT an ROI
	roiRequest := fmt.Sprintf("%snode/%s/%s/roi", server.WebAPIPath, uuid, data.DataName())
	server.TestHTTP(t, "POST", roiRequest, getSpansJSON(testSpans))

	// Get back the ROI
	returnedData := server.TestHTTP(t, "GET", roiRequest, nil)
	spans, err := putSpansJSON(returnedData)
	if err != nil {
		t.Errorf("Error on getting back JSON from roi GET: %v\n", err)
	}

	// Make sure the two are the same.
	if !reflect.DeepEqual(spans, testSpans) {
		t.Errorf("Bad PUT/GET ROI roundtrip\nOriginal:\n%s\nReturned:\n%s\n", testSpans, spans)
	}

	// Delete the ROI
	_ = server.TestHTTP(t, "DELETE", roiRequest, nil)

	// ROI should now be empty
	returnedData = server.TestHTTP(t, "GET", roiRequest, nil)
	if string(returnedData) != "[]" {
		t.Errorf("Bad ROI after ROI delete.  Should be [ ] got: %s\n", string(returnedData))
	}
}

func TestLabelblkRepoPersistence(t *testing.T) {
	tests.UseStore()
	defer tests.CloseStore()

	uuid, _ := initTestRepo()

	// Make labels and set various properties
	config := dvid.NewConfig()
	config.Set("BlockSize", "12,13,14")
	config.Set("VoxelSize", "1.1,2.8,11")
	config.Set("VoxelUnits", "microns,millimeters,nanometers")
	dataservice, err := datastore.NewData(uuid, labelsT, "mylabels", config)
	if err != nil {
		t.Errorf("Unable to create labels instance: %v\n", err)
	}
	labels, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast labels data service into Data\n")
	}
	oldData := *labels

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, labels); err != nil {
		t.Fatalf("Unable to save repo during labels persistence test: %v\n", err)
	}
	tests.CloseReopenStore()

	dataservice2, err := datastore.GetDataByUUID(uuid, "mylabels")
	if err != nil {
		t.Fatalf("Can't get labels instance from reloaded test db: %v\n", err)
	}
	labels2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not imageblk.Data\n")
	}
	if !oldData.Equals(labels2) {
		t.Errorf("Expected %v, got %v\n", oldData, *labels2)
	}
}

func TestFloat32RepoPersistence(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Make grayscale and set various properties
	config := dvid.NewConfig()
	config.Set("BlockSize", "12,13,14")
	config.Set("VoxelSize", "1.1,2.8,11")
	config.Set("VoxelUnits", "microns,millimeters,nanometers")
	dataservice, err := datastore.NewData(uuid, floatimgT, "floatimg", config)
	if err != nil {
		t.Errorf("Unable to create float32 instance: %s\n", err)
	}
	floatimg, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast float32 data service into Data\n")
	}
	oldData := *floatimg

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, floatimg); err != nil {
		t.Fatalf("Unable to save repo during floatimg persistence test: %v\n", err)
	}
	datastore.CloseReopenTest()

	dataservice2, err := datastore.GetDataByUUIDName(uuid, "floatimg")
	if err != nil {
		t.Fatalf("Can't get floatimg instance from reloaded test db: %v\n", err)
	}
	floatimg2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not imageblk.Data\n")
	}
	if !oldData.Equals(floatimg2) {
		t.Errorf("Expected %v, got %v\n", oldData, *floatimg2)
	}
}

func TestKeyvalueRoundTrip(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, versionID := initTestRepo()

	// Add data
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, kvtype, "roundtripper", config)
	if err != nil {
		t.Errorf("Error creating new keyvalue instance: %v\n", err)
	}
	kvdata, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Returned new data instance is not keyvalue.Data\n")
	}

	ctx := datastore.NewVersionedCtx(dataservice, versionID)

	keyStr := "testkey.-{}03`~| %@\x01"
	value := []byte("I like Japan and this is some unicode: \u65e5\u672c\u8a9e")

	if err = kvdata.PutData(ctx, keyStr, value); err != nil {
		t.Errorf("Could not put keyvalue data: %v\n", err)
	}

	retrieved, found, err := kvdata.GetData(ctx, keyStr)
	if err != nil {
		t.Fatalf("Could not get keyvalue data: %v\n", err)
	}
	if !found {
		t.Fatalf("Could not find put keyvalue\n")
	}
	if bytes.Compare(value, retrieved) != 0 {
		t.Errorf("keyvalue retrieved %q != put %q\n", string(retrieved), string(value))
	}
}

func TestMultichan16RepoPersistence(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	uuid, _ := initTestRepo()

	// Make labels and set various properties
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, dtype, "mymultichan16", config)
	if err != nil {
		t.Errorf("Unable to create multichan16 instance: %v\n", err)
	}
	mcdata, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast multichan16 data service into multichan16.Data\n")
	}
	oldData := *mcdata

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, mcdata); err != nil {
		t.Fatalf("Unable to save repo during multichan16 persistence test: %v\n", err)
	}
	datastore.CloseReopenTest()

	dataservice2, err := datastore.GetDataByUUIDName(uuid, "mymultichan16")
	if err != nil {
		t.Fatalf("Can't get multichan16 instance from reloaded test db: %v\n", err)
	}
	mcdata2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not multichan16.Data\n")
	}
	if !oldData.Equals(mcdata2) {
		t.Errorf("Expected %v, got %v\n", oldData, *mcdata2)
	}
}

func TestKeyvalueRepoPersistence(t *testing.T) {
	tests.UseStore()
	defer tests.CloseStore()

	uuid, _ := initTestRepo()

	// Make labels and set various properties
	config := dvid.NewConfig()
	dataservice, err := datastore.NewData(uuid, kvtype, "mykv", config)
	if err != nil {
		t.Errorf("Unable to create keyvalue instance: %v\n", err)
	}
	kvdata, ok := dataservice.(*Data)
	if !ok {
		t.Errorf("Can't cast keyvalue data service into keyvalue.Data\n")
	}
	oldData := *kvdata

	// Restart test datastore and see if datasets are still there.
	if err = datastore.SaveDataByUUID(uuid, kvdata); err != nil {
		t.Fatalf("Unable to save repo during keyvalue persistence test: %v\n", err)
	}
	tests.CloseReopenStore()

	dataservice2, err := datastore.GetDataByUUID(uuid, "mykv")
	if err != nil {
		t.Fatalf("Can't get keyvalue instance from reloaded test db: %v\n", err)
	}
	kvdata2, ok := dataservice2.(*Data)
	if !ok {
		t.Errorf("Returned new data instance 2 is not keyvalue.Data\n")
	}
	if !oldData.Equals(kvdata2) {
		t.Errorf("Expected %v, got %v\n", oldData, *kvdata2)
	}
}

func TestTileCheck(t *testing.T) {
	datastore.OpenTest()
	defer datastore.CloseTest()

	// Make source
	uuid, _ := initTestRepo()
	makeGrayscale(uuid, t, "grayscale")

	// Make imagetile and set various properties
	config := dvid.NewConfig()
	config.Set("Placeholder", "true")
	config.Set("Format", "jpg")
	config.Set("Source", "grayscale")
	tileservice, err := datastore.NewData(uuid, mstype, "myimagetile", config)
	if err != nil {
		t.Errorf("Unable to create imagetile instance: %v\n", err)
	}
	msdata, ok := tileservice.(*Data)
	if !ok {
		t.Fatalf("Can't cast imagetile data service into imagetile.Data\n")
	}

	// Store Metadata
	url := fmt.Sprintf("%snode/%s/myimagetile/metadata", server.WebAPIPath, uuid)
	server.TestHTTP(t, "POST", url, bytes.NewBufferString(testMetadata2))

	// Create the ROI
	_, err = datastore.NewData(uuid, roitype, "myroi", dvid.NewConfig())
	if err != nil {
		t.Errorf("Error creating new roi instance: %v\n", err)
	}

	// PUT an ROI
	roiRequest := fmt.Sprintf("%snode/%s/myroi/roi", server.WebAPIPath, uuid)
	server.TestHTTP(t, "POST", roiRequest, getSpansJSON(testSpans))

	// Create fake filter
	spec := fmt.Sprintf("roi:myroi,%s", uuid)
	f, err := msdata.NewFilter(storage.FilterSpec(spec))
	if err != nil {
		t.Errorf("Couldn't make filter: %v\n", err)
	}
	if f == nil {
		t.Fatalf("Couldn't detect myroi data instance\n")
	}

	// Check various key values for proper spatial checks.
	var tx, ty int32
	tx = (205 * 32) / 512
	ty = (101 * 32) / 512
	tile := dvid.ChunkPoint3d{tx, ty, 101 * 32}
	scale := Scaling(0)
	tkv := &storage.TKeyValue{K: NewTKey(tile, dvid.XY, scale)}
	skip, err := f.Check(tkv)
	if err != nil {
		t.Errorf("Error on Check of key %q: %v\n", tkv.K, err)
	}
	if skip {
		t.Errorf("Expected false skip, got %v for tile %s\n", skip, tile)
	}

	tile = dvid.ChunkPoint3d{tx, ty, 106 * 32}
	tkv = &storage.TKeyValue{K: NewTKey(tile, dvid.XY, scale)}
	skip, err = f.Check(tkv)
	if err != nil {
		t.Errorf("Error on Check of key %q: %v\n", tkv.K, err)
	}
	if !skip {
		t.Errorf("Expected true skip, got %v for tile %s\n", skip, tile)
	}

	tx = (205 * 32) / 512
	ty = (121 * 32) / 512
	tile = dvid.ChunkPoint3d{tx, ty, 101 * 32}
	tkv = &storage.TKeyValue{K: NewTKey(tile, dvid.XY, scale)}
	skip, err = f.Check(tkv)
	if err != nil {
		t.Errorf("Error on Check of key %q: %v\n", tkv.K, err)
	}
	if !skip {
		t.Errorf("Expected true skip, got %v for tile %s\n", skip, tile)
	}

	tx = (225 * 32) / 512
	ty = (101 * 32) / 512
	tile = dvid.ChunkPoint3d{tx, ty, 101 * 32}
	tkv = &storage.TKeyValue{K: NewTKey(tile, dvid.XY, scale)}
	skip, err = f.Check(tkv)
	if err != nil {
		t.Errorf("Error on Check of key %q: %v\n", tkv.K, err)
	}
	if !skip {
		t.Errorf("Expected true skip, got %v for tile %s\n", skip, tile)
	}

}