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

	uuid := createRepo(t)

	// Shouldn't be able to create branch on open node.
	branchReq := fmt.Sprintf("%snode/%s/branch", WebAPIPath, uuid)
	TestBadHTTP(t, "POST", branchReq, nil)

	// Commit it.
	payload := bytes.NewBufferString(`{"note": "This is my test commit", "log": ["line1", "line2", "some more stuff in a line"]}`)
	apiStr := fmt.Sprintf("%snode/%s/commit", WebAPIPath, uuid)
	TestHTTP(t, "POST", apiStr, payload)

	// Make sure committed nodes can only be read.
	// We shouldn't be able to write to log.
	payload = bytes.NewBufferString(`{"log": ["line1", "line2", "some more stuff in a line"]}`)
	apiStr = fmt.Sprintf("%snode/%s/log", WebAPIPath, uuid)
	TestBadHTTP(t, "POST", apiStr, payload)

	// Should be able to create branch now that we've committed parent.
	respData := TestHTTP(t, "POST", branchReq, nil)
	resp := struct {
		Child string `json:"child"`
	}{}
	if err := json.Unmarshal(respData, &resp); err != nil {
		t.Errorf("Expected 'child' JSON response.  Got %s\n", string(respData))
	}
	parent1 := dvid.UUID(resp.Child)

	// Create a sibling.
	respData = TestHTTP(t, "POST", branchReq, nil)
	if err := json.Unmarshal(respData, &resp); err != nil {
		t.Errorf("Expected 'child' JSON response.  Got %s\n", string(respData))
	}
	parent2 := dvid.UUID(resp.Child)

	// Commit both parents
	payload = bytes.NewBufferString(`{"note": "This is first parent"}`)
	apiStr = fmt.Sprintf("%snode/%s/commit", WebAPIPath, parent1)
	TestHTTP(t, "POST", apiStr, payload)

	payload = bytes.NewBufferString(`{"note": "This is second parent"}`)
	apiStr = fmt.Sprintf("%snode/%s/commit", WebAPIPath, parent2)
	TestHTTP(t, "POST", apiStr, payload)

	// Merge the two disjoint branches.
	mergeJSON := fmt.Sprintf(`{"mergeType": "conflict-free", "note": "This is my merged node", "parents": [%q, %q]}`, parent1[:7], parent2)
	payload = bytes.NewBufferString(mergeJSON)
	apiStr = fmt.Sprintf("%srepo/%s/merge", WebAPIPath, parent1)
	TestHTTP(t, "POST", apiStr, payload)
}

func TestDataGobEncoding(t *testing.T) {
	compression, _ := dvid.NewCompression(dvid.LZ4, dvid.DefaultCompression)
	data := &TestData{&Data{
		typename:    "testtype",
		typeurl:     "foo.bar.baz/testtype",
		typeversion: "1.0",
		id:          dvid.InstanceID(13),
		name:        "my fabulous data",
		rootUUID:    dvid.UUID("42"),
		dataUUID:    dvid.NewUUID(),
		compression: compression,
		checksum:    dvid.DefaultChecksum,
		syncData:    dvid.UUIDSet{"moo": struct{}{}, "bar": struct{}{}, "baz": struct{}{}},
	}}

	encoding, err := data.GobEncode()
	if err != nil {
		t.Fatalf("Couldn't Gob encode test data: %v\n", err)
	}
	data2 := &TestData{new(Data)}
	if err = data2.GobDecode(encoding); err != nil {
		t.Fatalf("Couldn't Gob decode test data: %v\n", err)
	}
	if !reflect.DeepEqual(data, data2) {
		t.Errorf("Bad Gob roundtrip:\nOriginal: %v\nReturned: %v\n", data, data2)
	}
}

// Lookup returns a mounted UUID Node.
func (MountDir) Lookup(name string, intr fs.Intr) (fs.Node, fuse.Error) {
	uuid := dvid.UUID(name) // We expect fully formed UUID, not partial.
	mount, found := fuseServer.mounts[uuid]
	if found {
		return &VersionDir{mount}, nil
	}
	return nil, fuse.ENOENT
}

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

	uuid := dvid.UUID(server.NewTestRepo(t))

	// Create new voxels instance with optional parameters
	name := "grayscale"
	metadata := fmt.Sprintf(`{
		"typename": "uint8blk",
		"dataname": %q,
		"blocksize": "64,43,28",
		"VoxelSize": "13.1, 14.2, 15.3",
		"VoxelUnits": "picometers,nanometers,microns"
	}`, name)
	apiStr := fmt.Sprintf("%srepo/%s/instance", server.WebAPIPath, uuid)
	server.TestHTTP(t, "POST", apiStr, bytes.NewBufferString(metadata))

	// Get metadata and make sure optional settings have been set.
	apiStr = fmt.Sprintf("%snode/%s/%s/info", server.WebAPIPath, uuid, name)
	result := server.TestHTTP(t, "GET", apiStr, nil)
	var parsed = struct {
		Base struct {
			TypeName, Name string
		}
		Extended struct {
			BlockSize  dvid.Point3d
			VoxelSize  dvid.NdFloat32
			VoxelUnits dvid.NdString
		}
	}{}
	if err := json.Unmarshal(result, &parsed); err != nil {
		t.Fatalf("Error parsing JSON response of new instance metadata: %v\n", err)
	}
	if parsed.Base.Name != name {
		t.Errorf("Parsed new instance has unexpected name: %s != %s (expected)\n",
			parsed.Base.Name, name)
	}
	if parsed.Base.TypeName != "uint8blk" {
		t.Errorf("Parsed new instance has unexpected type name: %s != uint8blk (expected)\n",
			parsed.Base.TypeName)
	}
	if !parsed.Extended.BlockSize.Equals(dvid.Point3d{64, 43, 28}) {
		t.Errorf("Bad block size in new uint8blk instance: %s\n", parsed.Extended.BlockSize)
	}
	if !parsed.Extended.VoxelSize.Equals(dvid.NdFloat32{13.1, 14.2, 15.3}) {
		t.Errorf("Bad voxel size in new uint8blk instance: %s\n", parsed.Extended.VoxelSize)
	}
	if parsed.Extended.VoxelUnits[0] != "picometers" {
		t.Errorf("Got %q for X voxel units, not picometers\n", parsed.Extended.VoxelUnits[0])
	}
	if parsed.Extended.VoxelUnits[1] != "nanometers" {
		t.Errorf("Got %q for X voxel units, not picometers\n", parsed.Extended.VoxelUnits[0])
	}
	if parsed.Extended.VoxelUnits[2] != "microns" {
		t.Errorf("Got %q for X voxel units, not picometers\n", parsed.Extended.VoxelUnits[0])
	}
}

func setupGroupcache(config GroupcacheConfig) error {
	if config.GB == 0 {
		return nil
	}
	var cacheBytes int64
	cacheBytes = int64(config.GB) << 30

	pool := groupcache.NewHTTPPool(config.Host)
	if pool != nil {
		dvid.Infof("Initializing groupcache with %d GB at %s...\n", config.GB, config.Host)
		manager.gcache.cache = groupcache.NewGroup("immutable", cacheBytes, groupcache.GetterFunc(
			func(c groupcache.Context, key string, dest groupcache.Sink) error {
				// Use KeyValueDB defined as context.
				gctx, ok := c.(GroupcacheCtx)
				if !ok {
					return fmt.Errorf("bad groupcache context: expected GroupcacheCtx, got %v", c)
				}

				// First four bytes of key is instance ID to isolate groupcache collisions.
				tk := TKey(key[4:])
				data, err := gctx.KeyValueDB.Get(gctx.Context, tk)
				if err != nil {
					return err
				}
				return dest.SetBytes(data)
			}))
		manager.gcache.supported = make(map[dvid.DataSpecifier]struct{})
		for _, dataspec := range config.Instances {
			name := strings.Trim(dataspec, "\"")
			parts := strings.Split(name, ":")
			switch len(parts) {
			case 2:
				dataid := dvid.GetDataSpecifier(dvid.InstanceName(parts[0]), dvid.UUID(parts[1]))
				manager.gcache.supported[dataid] = struct{}{}
			default:
				dvid.Errorf("bad data instance specification %q given for groupcache support in config file\n", dataspec)
			}
		}

		// If we have additional peers, add them and start a listener via the HTTP port.
		if len(config.Peers) > 0 {
			peers := []string{config.Host}
			peers = append(peers, config.Peers...)
			pool.Set(peers...)
			dvid.Infof("Groupcache configuration has %d peers in addition to local host.\n", len(config.Peers))
			dvid.Infof("Starting groupcache HTTP server on %s\n", config.Host)
			http.ListenAndServe(config.Host, http.HandlerFunc(pool.ServeHTTP))
		}
	}
	return nil
}

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

	apiStr := fmt.Sprintf("%srepos", server.WebAPIPath)
	r := server.TestHTTP(t, "POST", apiStr, nil)
	var jsonResp map[string]interface{}

	if err := json.Unmarshal(r, &jsonResp); err != nil {
		t.Fatalf("Unable to unmarshal repo creation response: %s\n", string(r))
	}
	v, ok := jsonResp["root"]
	if !ok {
		t.Fatalf("No 'root' metadata returned: %s\n", string(r))
	}
	uuidStr, ok := v.(string)
	if !ok {
		t.Fatalf("Couldn't cast returned 'root' data (%v) into string.\n", v)
	}
	uuid := dvid.UUID(uuidStr)

	// Shouldn't be able to create branch on open node.
	branchReq := fmt.Sprintf("%snode/%s/branch", server.WebAPIPath, uuid)
	server.TestBadHTTP(t, "POST", branchReq, nil)

	// Add a keyvalue instance.
	server.CreateTestInstance(t, uuid, "keyvalue", "mykv", dvid.Config{})

	// Commit it.
	payload := bytes.NewBufferString(`{"note": "This is my test commit", "log": ["line1", "line2", "some more stuff in a line"]}`)
	apiStr = fmt.Sprintf("%snode/%s/commit", server.WebAPIPath, uuid)
	server.TestHTTP(t, "POST", apiStr, payload)

	// Make sure committed nodes can only be read.
	// We shouldn't be able to write to keyvalue..
	keyReq := fmt.Sprintf("%snode/%s/mykv/key/foo", server.WebAPIPath, uuid)
	server.TestBadHTTP(t, "POST", keyReq, bytes.NewBufferString("some data"))

	// Should be able to create branch now that we've committed parent.
	respData := server.TestHTTP(t, "POST", branchReq, nil)
	resp := struct {
		Child dvid.UUID `json:"child"`
	}{}
	if err := json.Unmarshal(respData, &resp); err != nil {
		t.Errorf("Expected 'child' JSON response.  Got %s\n", string(respData))
	}

	// We should be able to write to that keyvalue now in the child.
	keyReq = fmt.Sprintf("%snode/%s/mykv/key/foo", server.WebAPIPath, resp.Child)
	server.TestHTTP(t, "POST", keyReq, bytes.NewBufferString("some data"))
}

func parseConfig(config dvid.StoreConfig) (path string, timeout time.Duration, owner string, collection dvid.UUID, err error) {
	c := config.GetAll()

	v, found := c["path"]
	if !found {
		err = fmt.Errorf("%q must be specified for kvautobus configuration", "path")
		return
	}
	var ok bool
	path, ok = v.(string)
	if !ok {
		err = fmt.Errorf("%q setting must be a string (%v)", "path", v)
		return
	}

	v, found = c["timeout"]
	if found {
		t, ok := v.(int64)
		if !ok {
			err = fmt.Errorf("%q setting must be an int64 for # seconds, not %s (%v)", "timeout", reflect.TypeOf(v), v)
			return
		}
		if t != 0 {
			timeout = time.Duration(t) * time.Second
		}
	}

	v, found = c["collection"]
	if !found {
		err = fmt.Errorf("kvautobus store must have collection specification for billing.")
		return
	}
	cstr, ok := v.(string)
	if !ok {
		err = fmt.Errorf("%q setting must be a string, not %s (%v)", "collection", reflect.TypeOf(v), v)
		return
	}
	collection = dvid.UUID(cstr)

	v, found = c["owner"]
	if !found {
		err = fmt.Errorf("kvautobus store must have owner specification for billing.")
		return
	}
	owner, ok = v.(string)
	if !ok {
		err = fmt.Errorf("%q setting must be a string, not %s (%v)", "owner", reflect.TypeOf(v), v)
		return
	}
	return
}

func TestRepoGobEncoding(t *testing.T) {
	uuid := dvid.UUID("19b87f38f873481b9f3ac688877dff0d")
	versionID := dvid.VersionID(23)
	repoID := dvid.RepoID(13)

	repo := newRepo(uuid, versionID, repoID, "foobar")
	repo.alias = "just some alias"
	repo.log = []string{
		"Did this",
		"Then that",
		"And the other thing",
	}
	repo.properties = map[string]interface{}{
		"foo": 42,
		"bar": "some string",
		"baz": []int{3, 9, 7},
	}

	encoding, err := repo.GobEncode()
	if err != nil {
		t.Fatalf("Could not encode repo: %v\n", err)
	}
	received := repoT{}
	if err = received.GobDecode(encoding); err != nil {
		t.Fatalf("Could not decode repo: %v\n", err)
	}

	// Did we serialize OK
	repo.dag = nil
	received.dag = nil
	if len(received.properties) != 3 {
		t.Errorf("Repo Gob messed up properties: %v\n", received.properties)
	}
	foo, ok := received.properties["foo"]
	if !ok || foo != 42 {
		t.Errorf("Repo Gob messed up properties: %v\n", received.properties)
	}
	bar, ok := received.properties["bar"]
	if !ok || bar != "some string" {
		t.Errorf("Repo Gob messed up properties: %v\n", received.properties)
	}
	baz, ok := received.properties["baz"]
	if !ok || !reflect.DeepEqual(baz, []int{3, 9, 7}) {
		t.Errorf("Repo Gob messed up properties: %v\n", received.properties)
	}
	repo.properties = nil
	received.properties = nil
	if !reflect.DeepEqual(*repo, received) {
		t.Fatalf("Repo Gob messed up:\nOriginal: %v\nReceived: %v\n", *repo, received)
	}
}

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

	uuidStr1 := "de305d5475b4431badb2eb6b9e546014"
	myuuid := dvid.UUID(uuidStr1)
	root, err := NewRepo("test repo", "test repo description", &myuuid, "")
	if err != nil {
		t.Fatal(err)
	}
	if root != myuuid {
		t.Errorf("Assigned root UUID %q != created root UUID %q\n", myuuid, root)
	}

	// Check if branches can also have assigned UUIDs
	if err := Commit(root, "root node", nil); err != nil {
		t.Fatal(err)
	}
	uuidStr2 := "8fa05d5475b4431badb2eb6b9e0123014"
	myuuid2 := dvid.UUID(uuidStr2)
	child, err := NewVersion(myuuid, "note describing uuid2", &myuuid2)
	if err != nil {
		t.Fatal(err)
	}
	if child != myuuid2 {
		t.Errorf("Assigned child UUID %q != created child UUID %q\n", myuuid2, child)
	}

	// Make sure we can lookup assigned UUIDs
	uuid, _, err := MatchingUUID(uuidStr1[:10])
	if err != nil {
		t.Errorf("Error matching UUID fragment %s: %v\n", uuidStr1[:10], err)
	}
	if uuid != myuuid {
		t.Errorf("Error getting back correct UUID %s from %s\n", myuuid, uuid)
	}

}

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

	uuid, _ := datastore.NewTestRepo()

	// Add data instances
	var config dvid.Config
	server.CreateTestInstance(t, uuid, "keyvalue", "foo", config)
	server.CreateTestInstance(t, uuid, "labelblk", "labels", config)
	server.CreateTestInstance(t, uuid, "roi", "someroi", config)

	// Reload the metadata
	apiStr := fmt.Sprintf("%sserver/reload-metadata", server.WebAPIPath)
	server.TestHTTP(t, "POST", apiStr, nil)

	// Make sure repo UUID still there
	jsonStr, err := datastore.MarshalJSON()
	if err != nil {
		t.Fatalf("can't get repos JSON: %v\n", err)
	}
	var jsonResp map[string](map[string]interface{})

	if err := json.Unmarshal(jsonStr, &jsonResp); err != nil {
		t.Fatalf("Unable to unmarshal repos info response: %s\n", jsonStr)
	}
	if len(jsonResp) != 1 {
		t.Errorf("reloaded repos had more than one repo: %v\n", jsonResp)
	}
	for k := range jsonResp {
		if dvid.UUID(k) != uuid {
			t.Fatalf("Expected uuid %s, got %s.  Full JSON:\n%v\n", uuid, k, jsonResp)
		}
	}

	// Make sure the data instances are still there.
	_, err = datastore.GetDataByUUIDName(uuid, "foo")
	if err != nil {
		t.Errorf("Couldn't get keyvalue data instance after reload\n")
	}
	_, err = datastore.GetDataByUUIDName(uuid, "labels")
	if err != nil {
		t.Errorf("Couldn't get labelblk data instance after reload\n")
	}
	_, err = datastore.GetDataByUUIDName(uuid, "someroi")
	if err != nil {
		t.Errorf("Couldn't get roi data instance after reload\n")
	}
}

func createRepo(t *testing.T) dvid.UUID {
	apiStr := fmt.Sprintf("%srepos", WebAPIPath)
	r := TestHTTP(t, "POST", apiStr, nil)
	var jsonResp map[string]interface{}

	if err := json.Unmarshal(r, &jsonResp); err != nil {
		t.Fatalf("Unable to unmarshal repo creation response: %s\n", string(r))
	}
	v, ok := jsonResp["root"]
	if !ok {
		t.Fatalf("No 'root' metadata returned: %s\n", string(r))
	}
	uuid, ok := v.(string)
	if !ok {
		t.Fatalf("Couldn't cast returned 'root' data (%v) into string.\n", v)
	}
	return dvid.UUID(uuid)
}

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

	apiStr := fmt.Sprintf("%srepos", server.WebAPIPath)
	r := server.TestHTTP(t, "POST", apiStr, nil)
	var jsonResp map[string]interface{}

	if err := json.Unmarshal(r, &jsonResp); err != nil {
		t.Fatalf("Unable to unmarshal repo creation response: %s\n", string(r))
	}
	v, ok := jsonResp["root"]
	if !ok {
		t.Fatalf("No 'root' metadata returned: %s\n", string(r))
	}
	uuidStr, ok := v.(string)
	if !ok {
		t.Fatalf("Couldn't cast returned 'root' data (%v) into string.\n", v)
	}
	uuid := dvid.UUID(uuidStr)

	// Add a data instance.
	var config dvid.Config
	server.CreateTestInstance(t, uuid, "keyvalue", "foo", config)

	// Make sure it exists.
	_, err := datastore.GetDataByUUID(uuid, "foo")
	if err != nil {
		t.Errorf("Couldn't create data instance 'foo'\n")
	}

	// Shouldn't be able to delete instance without "imsure"
	delReq := fmt.Sprintf("%srepo/%s/%s", server.WebAPIPath, uuid, "foo")
	server.TestBadHTTP(t, "DELETE", delReq, nil)
	delReq = fmt.Sprintf("%srepo/%s/%s?imsure=true", server.WebAPIPath, uuid, "foo")
	server.TestHTTP(t, "DELETE", delReq, nil)

	// Make sure it no longer exists.
	_, err = datastore.GetDataByUUID(uuid, "foo")
	if err == nil {
		t.Errorf("Shouldn't be able to access a deleted data instance 'foo'\n")
	}
}

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

	uuid := dvid.UUID(server.NewTestRepo(t))
	if len(uuid) < 5 {
		t.Fatalf("Bad root UUID for new repo: %s\n", uuid)
	}

	// Create a labelblk instance
	vol := labelVol{
		size:      dvid.Point3d{5, 5, 5}, // in blocks
		blockSize: dvid.Point3d{32, 32, 32},
		offset:    dvid.Point3d{32, 64, 96},
	}
	vol.postLabelVolume(t, "labels", uuid)

	// TODO -- Test syncing across labelblk, labelvol, labelsz.
}

func makeTestVersions(t *testing.T) {
	root, err := NewRepo("test repo", "test repo description", nil, "")
	if err != nil {
		t.Fatal(err)
	}

	if err := Commit(root, "root node", nil); err != nil {
		t.Fatal(err)
	}

	child1, err := NewVersion(root, "note describing child 1", nil)
	if err != nil {
		t.Fatal(err)
	}

	if err := Commit(child1, "child 1", nil); err != nil {
		t.Fatal(err)
	}

	// Test ability to set UUID of child
	assignedUUID := dvid.UUID("0c8bc973dba74729880dd1bdfd8d0c5e")
	child2, err := NewVersion(root, "note describing child 2", &assignedUUID)
	if err != nil {
		t.Fatal(err)
	}

	log2 := []string{"This is line 1 of log", "This is line 2 of log", "Last line for multiline log"}
	if err := Commit(child2, "child 2 assigned", log2); err != nil {
		t.Fatal(err)
	}

	// Make uncommitted child 3
	child3, err := NewVersion(root, "note describing child 3", nil)
	if err != nil {
		t.Fatal(err)
	}
	nodelog := []string{`My first node-level log line.!(;#)}`, "Second line is here!!!"}
	if err := AddToNodeLog(child3, nodelog); err != nil {
		t.Fatal(err)
	}
}

func TestParseConfig(t *testing.T) {
	var tc tomlConfig
	if _, err := toml.Decode(testConfig, &tc); err != nil {
		t.Fatalf("Could not decode TOML config: %v\n", err)
	}
	sc, ok := tc.Store["kvautobus"]
	if !ok {
		t.Fatalf("Couldn't find kvautobus config in test\n")
	}
	var config dvid.Config
	config.SetAll(sc)
	kvconfig := dvid.StoreConfig{
		Config: config,
		Engine: "kvautobus",
	}
	path, timeout, owner, collection, err := parseConfig(kvconfig)
	if err != nil {
		t.Errorf("Error parsing kvautobus config: %v\n", err)
	}

	if path != "http://tem-dvid.int.janelia.org:9000" {
		t.Errorf("Bad parsing of kvautobus config.  Path = %s, not http://tem-dvid.int.janelia.org:9000", path)
	}

	if timeout != time.Duration(30)*time.Second {
		t.Errorf("Expected parsing of kvautobus config: timeout = 30 * time.Second, got %d\n", timeout)
	}

	if owner != "flyEM" {
		t.Errorf("expected owner for kvautobus to be %q, got %q\n", "flyEM", owner)
	}

	if collection != dvid.UUID("99ef22cd85f143f58a623bd22aad0ef7") {
		t.Errorf("expected collection for kvautobus to be 99ef22cd85f143f58a623bd22aad0ef7, got %s\n", collection)
	}
}

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

	uuid := createRepo(t)

	// Shouldn't be able to create branch on open node.
	branchReq := fmt.Sprintf("%snode/%s/branch", WebAPIPath, uuid)
	TestBadHTTP(t, "POST", branchReq, nil)

	// Commit it.
	payload := bytes.NewBufferString(`{"note": "This is my test commit", "log": ["line1", "line2", "some more stuff in a line"]}`)
	apiStr := fmt.Sprintf("%snode/%s/commit", WebAPIPath, uuid)
	TestHTTP(t, "POST", apiStr, payload)

	// Make sure committed nodes can only be read.
	// We shouldn't be able to write to log.
	payload = bytes.NewBufferString(`{"log": ["line1", "line2", "some more stuff in a line"]}`)
	apiStr = fmt.Sprintf("%snode/%s/log", WebAPIPath, uuid)
	TestBadHTTP(t, "POST", apiStr, payload)

	// Should be able to create branch now that we've committed parent.
	respData := TestHTTP(t, "POST", branchReq, nil)
	resp := struct {
		Child string `json:"child"`
	}{}
	if err := json.Unmarshal(respData, &resp); err != nil {
		t.Errorf("Expected 'child' JSON response.  Got %s\n", string(respData))
	}
	parent1 := dvid.UUID(resp.Child)

	// Create a sibling.
	respData = TestHTTP(t, "POST", branchReq, nil)
	if err := json.Unmarshal(respData, &resp); err != nil {
		t.Errorf("Expected 'child' JSON response.  Got %s\n", string(respData))
	}
	parent2 := dvid.UUID(resp.Child)

	// Commit both parents
	payload = bytes.NewBufferString(`{"note": "This is first parent"}`)
	apiStr = fmt.Sprintf("%snode/%s/commit", WebAPIPath, parent1)
	TestHTTP(t, "POST", apiStr, payload)

	payload = bytes.NewBufferString(`{"note": "This is second parent"}`)
	apiStr = fmt.Sprintf("%snode/%s/commit", WebAPIPath, parent2)
	TestHTTP(t, "POST", apiStr, payload)

	// Merge the two disjoint branches.
	mergeJSON := fmt.Sprintf(`{"mergeType": "conflict-free", "note": "This is my merged node", "parents": [%q, %q]}`, parent1[:7], parent2)
	payload = bytes.NewBufferString(mergeJSON)
	apiStr = fmt.Sprintf("%srepo/%s/merge", WebAPIPath, parent1)
	TestHTTP(t, "POST", apiStr, payload)

	// Get root version to check after delete repo.
	rootV, err := datastore.VersionFromUUID(uuid)
	if err != nil {
		t.Errorf("Got unexpected error on getting version from root UUID: %v\n", err)
	}

	// Delete the entire repo including all branches.
	apiStr = fmt.Sprintf("%srepo/%s", WebAPIPath, parent2)
	TestBadHTTP(t, "DELETE", apiStr, nil) // Requires query string
	apiStr = fmt.Sprintf("%srepo/%s?imsure=true", WebAPIPath, parent2)
	TestHTTP(t, "DELETE", apiStr, nil) // Requires query string

	// Make sure none of the repo is still accessible.
	jsonResp, err := datastore.GetRepoJSON(uuid)
	if err == nil {
		t.Errorf("Expected invalid UUID after repo delete but got json back: %s\n", jsonResp)
	}
	if err != datastore.ErrInvalidUUID {
		t.Errorf("Expected invalid UUID after repo delete but got unexpected error: %v\n", err)
	}
	_, err = datastore.VersionFromUUID(uuid)
	if err != datastore.ErrInvalidUUID {
		t.Errorf("Expected invalid rot UUID after repo delete but got unexpected error: %v\n", err)
	}
	_, err = datastore.VersionFromUUID(parent1)
	if err != datastore.ErrInvalidUUID {
		t.Errorf("Expected invalid UUID for 1st parent after repo delete but got unexpected error: %v\n", err)
	}
	_, err = datastore.VersionFromUUID(parent2)
	if err != datastore.ErrInvalidUUID {
		t.Errorf("Expected invalid UUID for 2nd parent after repo delete but got unexpected error: %v\n", err)
	}
	_, err = datastore.UUIDFromVersion(rootV)
	if err != datastore.ErrInvalidVersion {
		t.Errorf("Expected invalid version id for root after repo delete but got unexpected error: %v\n", err)
	}
}

// switchboard for remote command execution
func handleCommand(cmd *datastore.Request) (reply *datastore.Response, err error) {
	if cmd.Name() == "" {
		err = fmt.Errorf("Server error: got empty command!")
		return
	}
	reply = new(datastore.Response)

	switch cmd.Name() {

	case "help":
		reply.Text = fmt.Sprintf(RPCHelpMessage, config.RPCAddress(), config.HTTPAddress())

	case "shutdown":
		dvid.Infof("DVID server halting due to 'shutdown' command.")
		reply.Text = fmt.Sprintf("DVID server at %s is being shutdown...\n", config.RPCAddress())
		// launch goroutine shutdown so we can concurrently return shutdown message to client.
		go Shutdown()

	case "types":
		if len(cmd.Command) == 1 {
			text := "\nData Types within this DVID Server\n"
			text += "----------------------------------\n"
			var mapTypes map[dvid.URLString]datastore.TypeService
			if mapTypes, err = datastore.Types(); err != nil {
				err = fmt.Errorf("Error trying to retrieve data types within this DVID server!")
				return
			}
			for url, typeservice := range mapTypes {
				text += fmt.Sprintf("%-20s %s\n", typeservice.GetTypeName(), url)
			}
			reply.Text = text
		} else {
			if len(cmd.Command) != 3 || cmd.Command[2] != "help" {
				err = fmt.Errorf("Unknown types command: %q", cmd.Command)
				return
			}
			var typename string
			var typeservice datastore.TypeService
			cmd.CommandArgs(1, &typename)
			if typeservice, err = datastore.TypeServiceByName(dvid.TypeString(typename)); err != nil {
				return
			}
			reply.Text = typeservice.Help()
		}

	case "repos":
		var subcommand string
		cmd.CommandArgs(1, &subcommand)

		switch subcommand {
		case "new":
			var alias, description string
			cmd.CommandArgs(2, &alias, &description)

			config := cmd.Settings()
			var uuidStr, passcode string
			var found bool
			if uuidStr, found, err = config.GetString("uuid"); err != nil {
				return
			}
			var assign *dvid.UUID
			if !found {
				assign = nil
			} else {
				uuid := dvid.UUID(uuidStr)
				assign = &uuid
			}
			if passcode, found, err = config.GetString("passcode"); err != nil {
				return
			}
			var root dvid.UUID
			root, err = datastore.NewRepo(alias, description, assign, passcode)
			if err != nil {
				return
			}
			if err = datastore.SetRepoAlias(root, alias); err != nil {
				return
			}
			if err = datastore.SetRepoDescription(root, description); err != nil {
				return
			}
			reply.Text = fmt.Sprintf("New repo %q created with head node %s\n", alias, root)

		case "delete":
			var uuidStr, passcode string
			cmd.CommandArgs(2, &uuidStr, &passcode)

			var uuid dvid.UUID
			if uuid, _, err = datastore.MatchingUUID(uuidStr); err != nil {
				return
			}
			if err = datastore.DeleteRepo(uuid, passcode); err != nil {
				return
			}
			reply.Text = fmt.Sprintf("Started deletion of repo %s.\n", uuid)

		default:
			err = fmt.Errorf("Unknown repos command: %q", subcommand)
			return
//.........这里部分代码省略.........

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

	uuid := dvid.UUID(server.NewTestRepo(t))
	if len(uuid) < 5 {
		t.Fatalf("Bad root UUID for new repo: %s\n", uuid)
	}

	// Create a labelblk instance
	server.CreateTestInstance(t, uuid, "labelblk", "labels", dvid.Config{})

	vol := labelVol{
		size:      dvid.Point3d{5, 5, 5}, // in blocks
		blockSize: dvid.Point3d{32, 32, 32},
		offset:    dvid.Point3d{32, 64, 96},
		name:      "labels",
	}
	vol.postLabelVolume(t, uuid, "", "", 0)

	// Repost the label volume 3 more times with increasing starting values.
	vol.postLabelVolume(t, uuid, "", "", 2100)
	vol.postLabelVolume(t, uuid, "", "", 8176)
	vol.postLabelVolume(t, uuid, "", "", 16623)

	vol.testSlices(t, uuid)

	// Try to post last volume concurrently 3x and then check result.
	wg := new(sync.WaitGroup)
	wg.Add(3)
	go func() {
		vol.postLabelVolume(t, uuid, "", "", 16623)
		wg.Done()
	}()
	go func() {
		vol.postLabelVolume(t, uuid, "", "", 16623)
		wg.Done()
	}()
	go func() {
		vol.postLabelVolume(t, uuid, "", "", 16623)
		wg.Done()
	}()
	wg.Wait()
	vol.testGetLabelVolume(t, uuid, "", "")

	// Try concurrent write of disjoint subvolumes.
	vol2 := labelVol{
		size:      dvid.Point3d{5, 5, 5}, // in blocks
		blockSize: dvid.Point3d{32, 32, 32},
		offset:    dvid.Point3d{192, 64, 96},
		name:      "labels",
	}
	vol3 := labelVol{
		size:      dvid.Point3d{5, 5, 5}, // in blocks
		blockSize: dvid.Point3d{32, 32, 32},
		offset:    dvid.Point3d{192, 224, 96},
		name:      "labels",
	}
	vol4 := labelVol{
		size:      dvid.Point3d{5, 5, 5}, // in blocks
		blockSize: dvid.Point3d{32, 32, 32},
		offset:    dvid.Point3d{32, 224, 96},
		name:      "labels",
	}

	wg.Add(3)
	go func() {
		vol2.postLabelVolume(t, uuid, "lz4", "", 4000)
		wg.Done()
	}()
	go func() {
		vol3.postLabelVolume(t, uuid, "lz4", "", 8000)
		wg.Done()
	}()
	go func() {
		vol4.postLabelVolume(t, uuid, "lz4", "", 1200)
		wg.Done()
	}()
	wg.Wait()
	vol.testGetLabelVolume(t, uuid, "", "")
	vol2.testGetLabelVolume(t, uuid, "", "")
	vol3.testGetLabelVolume(t, uuid, "", "")
	vol4.testGetLabelVolume(t, uuid, "", "")

	// Verify various GET 3d volume with compressions and no ROI.
	vol.testGetLabelVolume(t, uuid, "", "")
	vol.testGetLabelVolume(t, uuid, "lz4", "")
	vol.testGetLabelVolume(t, uuid, "gzip", "")

	// Create a new ROI instance.
	roiName := "myroi"
	server.CreateTestInstance(t, uuid, "roi", roiName, dvid.Config{})

	// Add ROI data
	apiStr := fmt.Sprintf("%snode/%s/%s/roi", server.WebAPIPath, uuid, roiName)
	server.TestHTTP(t, "POST", apiStr, bytes.NewBufferString(labelsJSON()))

	// Post updated labels without ROI and make sure it returns those values.
	var labelNoROI uint64 = 20000
	vol.postLabelVolume(t, uuid, "", "", labelNoROI)
//.........这里部分代码省略.........

// Do acts as a switchboard for remote command execution
func (c *RPCConnection) Do(cmd datastore.Request, reply *datastore.Response) error {
	if reply == nil {
		dvid.Debugf("reply is nil coming in!\n")
		return nil
	}
	if cmd.Name() == "" {
		return fmt.Errorf("Server error: got empty command!")
	}

	switch cmd.Name() {

	case "help":
		reply.Text = fmt.Sprintf(RPCHelpMessage, config.RPCAddress(), config.HTTPAddress())

	case "shutdown":
		Shutdown()
		// Make this process shutdown in a second to allow time for RPC to finish.
		// TODO -- Better way to do this?
		log.Printf("DVID server halted due to 'shutdown' command.")
		reply.Text = fmt.Sprintf("DVID server at %s has been halted.\n", config.RPCAddress())
		go func() {
			time.Sleep(1 * time.Second)
			os.Exit(0)
		}()

	case "types":
		if len(cmd.Command) == 1 {
			text := "\nData Types within this DVID Server\n"
			text += "----------------------------------\n"
			mapTypes, err := datastore.Types()
			if err != nil {
				return fmt.Errorf("Error trying to retrieve data types within this DVID server!")
			}
			for url, typeservice := range mapTypes {
				text += fmt.Sprintf("%-20s %s\n", typeservice.GetTypeName(), url)
			}
			reply.Text = text
		} else {
			if len(cmd.Command) != 3 || cmd.Command[2] != "help" {
				return fmt.Errorf("Unknown types command: %q", cmd.Command)
			}
			var typename string
			cmd.CommandArgs(1, &typename)
			typeservice, err := datastore.TypeServiceByName(dvid.TypeString(typename))
			if err != nil {
				return err
			}
			reply.Text = typeservice.Help()
		}

	case "repos":
		var subcommand, alias, description, uuidStr string
		cmd.CommandArgs(1, &subcommand, &alias, &description, &uuidStr)
		switch subcommand {
		case "new":
			var assign *dvid.UUID
			if uuidStr == "" {
				assign = nil
			} else {
				u := dvid.UUID(uuidStr)
				assign = &u
			}
			root, err := datastore.NewRepo(alias, description, assign)
			if err != nil {
				return err
			}
			if err := datastore.SetRepoAlias(root, alias); err != nil {
				return err
			}
			if err := datastore.SetRepoDescription(root, description); err != nil {
				return err
			}
			reply.Text = fmt.Sprintf("New repo %q created with head node %s\n", alias, root)
		default:
			return fmt.Errorf("Unknown repos command: %q", subcommand)
		}

	case "repo":
		var uuidStr, subcommand string
		cmd.CommandArgs(1, &uuidStr, &subcommand)
		uuid, _, err := datastore.MatchingUUID(uuidStr)
		if err != nil {
			return err
		}

		switch subcommand {
		case "new":
			var typename, dataname string
			cmd.CommandArgs(3, &typename, &dataname)

			// Get TypeService
			typeservice, err := datastore.TypeServiceByName(dvid.TypeString(typename))
			if err != nil {
				return err
			}

			// Create new data
			config := cmd.Settings()
			_, err = datastore.NewData(uuid, typeservice, dvid.InstanceName(dataname), config)
//.........这里部分代码省略.........

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

	apiStr := fmt.Sprintf("%srepos", server.WebAPIPath)
	r := server.TestHTTP(t, "POST", apiStr, nil)
	var jsonResp map[string]interface{}

	if err := json.Unmarshal(r, &jsonResp); err != nil {
		t.Fatalf("Unable to unmarshal repo creation response: %s\n", string(r))
	}
	v, ok := jsonResp["root"]
	if !ok {
		t.Fatalf("No 'root' metadata returned: %s\n", string(r))
	}
	uuidStr, ok := v.(string)
	if !ok {
		t.Fatalf("Couldn't cast returned 'root' data (%v) into string.\n", v)
	}
	uuid := dvid.UUID(uuidStr)

	// Shouldn't be able to create branch on open node.
	branchReq := fmt.Sprintf("%snode/%s/branch", server.WebAPIPath, uuid)
	server.TestBadHTTP(t, "POST", branchReq, nil)

	// Add a keyvalue instance.
	server.CreateTestInstance(t, uuid, "keyvalue", "mykv", dvid.Config{})

	// Commit it.
	payload := bytes.NewBufferString(`{"note": "This is my test commit", "log": ["line1", "line2", "some more stuff in a line"]}`)
	apiStr = fmt.Sprintf("%snode/%s/commit", server.WebAPIPath, uuid)
	server.TestHTTP(t, "POST", apiStr, payload)

	// Make sure committed nodes can only be read.
	// We shouldn't be able to write to keyvalue.
	keyReq := fmt.Sprintf("%snode/%s/mykv/key/foo", server.WebAPIPath, uuid)
	server.TestBadHTTP(t, "POST", keyReq, bytes.NewBufferString("some data"))

	// Create a ROI with an immutable POST request via ptquery.  Should be able to still POST to it.
	server.CreateTestInstance(t, uuid, "roi", "myroi",