func TestUnmarshalNext(t *testing.T) {
	// We only need to check against a few, not all of them.
	tests := unmarshalingTests[:5]

	// Create a buffer with many concatenated JSON objects.
	var b bytes.Buffer
	for _, tt := range tests {
		b.WriteString(tt.json)
	}

	dec := json.NewDecoder(&b)
	for _, tt := range tests {
		// Make a new instance of the type of our expected object.
		p := reflect.New(reflect.TypeOf(tt.pb).Elem()).Interface().(proto.Message)

		err := tt.unmarshaler.UnmarshalNext(dec, p)
		if err != nil {
			t.Errorf("%s: %v", tt.desc, err)
			continue
		}

		// For easier diffs, compare text strings of the protos.
		exp := proto.MarshalTextString(tt.pb)
		act := proto.MarshalTextString(p)
		if string(exp) != string(act) {
			t.Errorf("%s: got [%s] want [%s]", tt.desc, act, exp)
		}
	}

	p := &pb.Simple{}
	err := new(Unmarshaler).UnmarshalNext(dec, p)
	if err != io.EOF {
		t.Errorf("eof: got %v, expected io.EOF", err)
	}
}

func testGetSrvKeyspace(t *testing.T, conn *vtgateconn.VTGateConn) {
	want := &topodatapb.SrvKeyspace{
		Partitions: []*topodatapb.SrvKeyspace_KeyspacePartition{
			&topodatapb.SrvKeyspace_KeyspacePartition{
				ServedType: topodatapb.TabletType_REPLICA,
				ShardReferences: []*topodatapb.ShardReference{
					&topodatapb.ShardReference{
						Name: "shard0",
						KeyRange: &topodatapb.KeyRange{
							Start: []byte{0x40, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
							End:   []byte{0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
						},
					},
				},
			},
		},
		ShardingColumnName: "sharding_column_name",
		ShardingColumnType: topodatapb.KeyspaceIdType_UINT64,
		ServedFrom: []*topodatapb.SrvKeyspace_ServedFrom{
			&topodatapb.SrvKeyspace_ServedFrom{
				TabletType: topodatapb.TabletType_MASTER,
				Keyspace:   "other_keyspace",
			},
		},
		SplitShardCount: 128,
	}
	got, err := conn.GetSrvKeyspace(context.Background(), "big")
	if err != nil {
		t.Fatalf("GetSrvKeyspace error: %v", err)
	}
	if !proto.Equal(got, want) {
		t.Errorf("GetSrvKeyspace() = %v, want %v", proto.MarshalTextString(got), proto.MarshalTextString(want))
	}
}

func (_ singleTaskDecoder) DecodeToText(data []byte) (string, error) {
	var task single.Task
	var profileExtension single.ProfileExtension
	err := proto.Unmarshal(data, &task)
	if err != nil {
		return "", err
	}
	if task.Profile != nil && task.Profile.Extension != nil &&
		task.Profile.Extension.TypeUrl ==
			"type.googleapis.com/bacs.problem.single.ProfileExtension" {
		ext := task.Profile.Extension
		task.Profile = nil
		err = proto.Unmarshal(ext.Value, &profileExtension)
		if err != nil {
			return "", err
		}
	}
	text := proto.MarshalTextString(&task)
	profileExtensionText := proto.MarshalTextString(&profileExtension)
	if profileExtensionText != "" {
		text += "\n"
		text += profileExtensionText
	}
	return text, nil
}

func TestProtoCopy(t *testing.T) {
	person := InitData("ronaflx", 195936, "[email protected]")
	copyPerson := person
	if !proto.Equal(person, copyPerson) {
		t.Errorf("expect:\n%s\nactual:\n%s\n", proto.MarshalTextString(person), proto.MarshalTextString(
			copyPerson))
	}
	msg := proto.Clone(person)
	if !proto.Equal(person, msg) {
		t.Errorf("expect:\n%s\nactual:\n%s\n", proto.MarshalTextString(person), proto.MarshalTextString(
			msg))
	}
}

func TestHorizontalReshardingTaskEmittedTasks(t *testing.T) {
	reshardingTask := &HorizontalReshardingTask{}

	parameters := map[string]string{
		// Required parameters.
		"keyspace":          "test_keyspace",
		"source_shard_list": "10-20",
		"dest_shard_list":   "10-18,18-20",
		"vtctld_endpoint":   "localhost:15000",
		"vtworker_endpoint": "localhost:15001",
		// Optional parameters.
		"exclude_tables":             "unrelated1,unrelated2",
		"min_healthy_rdonly_tablets": "1",
	}

	err := validateParameters(reshardingTask, parameters)
	if err != nil {
		t.Fatalf("Not all required parameters were specified: %v", err)
	}

	newTaskContainers, _, _ := reshardingTask.Run(parameters)

	// TODO(mberlin): Check emitted tasks against expected output.
	for _, tc := range newTaskContainers {
		t.Logf("new tasks: %v", proto.MarshalTextString(tc))
	}
}

func TestSpan(t *testing.T) {
	const input = `package main

import "fmt"
func main() { fmt.Println("Hello, world") }`

	unit, digest := oneFileCompilation("main.go", "main", input)
	fetcher := memFetcher{digest: input}
	pi, err := Resolve(unit, fetcher, nil)
	if err != nil {
		t.Fatalf("Resolve failed: %v\nInput unit:\n%s", err, proto.MarshalTextString(unit))
	}

	tests := []struct {
		key      func(*ast.File) ast.Node // return a node to compute a span for
		pos, end int                      // the expected span for the node
	}{
		{func(*ast.File) ast.Node { return nil }, -1, -1},                  // invalid node
		{func(*ast.File) ast.Node { return fakeNode{0, 2} }, -1, -1},       // invalid pos
		{func(*ast.File) ast.Node { return fakeNode{5, 0} }, 4, 4},         // invalid end
		{func(f *ast.File) ast.Node { return f.Name }, 8, 12},              // main
		{func(f *ast.File) ast.Node { return f.Imports[0].Path }, 21, 26},  // "fmt"
		{func(f *ast.File) ast.Node { return f.Decls[0] }, 14, 26},         // import "fmt"
		{func(f *ast.File) ast.Node { return f.Decls[1] }, 27, len(input)}, // func main() { ... }
	}
	for _, test := range tests {
		node := test.key(pi.Files[0])
		pos, end := pi.Span(node)
		if pos != test.pos || end != test.end {
			t.Errorf("Span(%v): got pos=%d, end=%v; want pos=%d, end=%d", node, pos, end, test.pos, test.end)
		}
	}
}

func TestTaskEmitsTaskWhichCannotBeInstantiated(t *testing.T) {
	scheduler := newTestScheduler(t)
	defer scheduler.ShutdownAndWait()
	scheduler.setTaskCreator(func(taskName string) Task {
		// TaskCreator which doesn't know TestingEchoTask (but emitted by TestingEmitEchoTask).
		switch taskName {
		case "TestingEmitEchoTask":
			return &TestingEmitEchoTask{}
		default:
			return nil
		}
	})
	scheduler.registerClusterOperation("TestingEmitEchoTask")

	scheduler.Run()

	enqueueRequest := &automationpb.EnqueueClusterOperationRequest{
		Name: "TestingEmitEchoTask",
		Parameters: map[string]string{
			"echo_text": "to be emitted task should fail to instantiate",
		},
	}
	enqueueResponse, err := scheduler.EnqueueClusterOperation(context.Background(), enqueueRequest)
	if err != nil {
		t.Fatalf("Failed to start cluster operation. Request: %v Error: %v", enqueueRequest, err)
	}

	details := waitForClusterOperation(t, scheduler, enqueueResponse.Id, "emitted TestingEchoTask", "no implementation found for: TestingEchoTask")
	if len(details.SerialTasks) != 1 {
		t.Errorf("A task has been emitted, but it shouldn't. Details:\n%v", proto.MarshalTextString(details))
	}
}

// Analyze will determine which analyzers to run and call them as appropriate. If necessary, it will
// also modify the context before calling the analyzers. It recovers from all analyzer panics with a
// note that the analyzer failed.
func (s analyzerService) Analyze(ctx server.Context, in *rpcpb.AnalyzeRequest) (resp *rpcpb.AnalyzeResponse, err error) {
	resp = new(rpcpb.AnalyzeResponse)

	log.Printf("called with: %v", proto.MarshalTextString(in))
	log.Print("starting analyzing")
	var nts []*notepb.Note
	var errs []*rpcpb.AnalysisFailure

	defer func() {
		resp.Note = nts
		resp.Failure = errs
	}()

	orgDir, restore, err := file.ChangeDir(*in.ShipshapeContext.RepoRoot)
	if err != nil {
		log.Printf("Internal error before analyzing: %v", err)
		appendFailure(&errs, "InternalFailure", err)
		return resp, err
	}
	defer func() {
		if err := restore(); err != nil {
			log.Printf("could not return back into %s from %s: %v", orgDir, *in.ShipshapeContext.RepoRoot, err)
		}
	}()

	reqCats := strset.New(in.Category...)
	for _, a := range s.analyzers {
		if reqCats.Contains(a.Category()) {
			runAnalyzer(a, in.ShipshapeContext, &nts, &errs)
		}
	}
	log.Printf("finished analyzing, sending back %d notes and %d errors", len(nts), len(errs))
	return resp, nil
}

// NewEncoder returns a new encoder based on content type negotiation.
func NewEncoder(w io.Writer, format Format) Encoder {
	switch format {
	case FmtProtoDelim:
		return encoder(func(v *dto.MetricFamily) error {
			_, err := pbutil.WriteDelimited(w, v)
			return err
		})
	case FmtProtoCompact:
		return encoder(func(v *dto.MetricFamily) error {
			_, err := fmt.Fprintln(w, v.String())
			return err
		})
	case FmtProtoText:
		return encoder(func(v *dto.MetricFamily) error {
			_, err := fmt.Fprintln(w, proto.MarshalTextString(v))
			return err
		})
	case FmtText:
		return encoder(func(v *dto.MetricFamily) error {
			_, err := MetricFamilyToText(w, v)
			return err
		})
	}
	panic("expfmt.NewEncoder: unknown format")
}

func (_ brokerStatusDecoder) DecodeToText(data []byte) (string, error) {
	var status rabbit.RabbitStatus
	err := proto.Unmarshal(data, &status)
	if err != nil {
		return "", err
	}
	return proto.MarshalTextString(&status), nil
}

func ExampleConstHistogram() {
	desc := prometheus.NewDesc(
		"http_request_duration_seconds",
		"A histogram of the HTTP request durations.",
		[]string{"code", "method"},
		prometheus.Labels{"owner": "example"},
	)

	// Create a constant histogram from values we got from a 3rd party telemetry system.
	h := prometheus.MustNewConstHistogram(
		desc,
		4711, 403.34,
		map[float64]uint64{25: 121, 50: 2403, 100: 3221, 200: 4233},
		"200", "get",
	)

	// Just for demonstration, let's check the state of the histogram by
	// (ab)using its Write method (which is usually only used by Prometheus
	// internally).
	metric := &dto.Metric{}
	h.Write(metric)
	fmt.Println(proto.MarshalTextString(metric))

	// Output:
	// label: <
	//   name: "code"
	//   value: "200"
	// >
	// label: <
	//   name: "method"
	//   value: "get"
	// >
	// label: <
	//   name: "owner"
	//   value: "example"
	// >
	// histogram: <
	//   sample_count: 4711
	//   sample_sum: 403.34
	//   bucket: <
	//     cumulative_count: 121
	//     upper_bound: 25
	//   >
	//   bucket: <
	//     cumulative_count: 2403
	//     upper_bound: 50
	//   >
	//   bucket: <
	//     cumulative_count: 3221
	//     upper_bound: 100
	//   >
	//   bucket: <
	//     cumulative_count: 4233
	//     upper_bound: 200
	//   >
	// >
}

func main() {
	flag.Parse()
	logger := xlog.NewPlainLogger()

	const usage = "usage: rcall service::[email protected] [request_contents]"

	if len(flag.Args()) != 1 && len(flag.Args()) != 2 {
		logger.Fatal(usage)
	}

	svcAndAddr := strings.Split(flag.Arg(0), "@")
	if len(svcAndAddr) != 2 {
		logger.Fatal(usage)
	}
	svcAndMethod := strings.Split(svcAndAddr[0], "::")
	if len(svcAndMethod) != 2 {
		logger.Fatal(usage)
	}

	ctrl, err := rpc.NewController(rpc.Config{
		Logger: xlog.NewNilLogger(),
	})
	if err != nil {
		logger.Fatalf("Failed to create RPC controller: %s", err)
	}

	client, err := ctrl.NewClient(rpc.ClientOptions{
		ServiceName:  svcAndMethod[0],
		ServiceAddr:  svcAndAddr[1],
		ConnPoolSize: 1,
		Retry:        rpc.DefaultDialRetry,
	})
	if err != nil {
		logger.Fatalf("Failed to create RPC client: %s", err)
	}

	var requestPB *string
	if len(flag.Args()) == 2 {
		s := flag.Arg(1)
		requestPB = &s
	}
	parentCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	ctx := &rpc.ClientContext{Context: parentCtx}
	responsePB, err := client.CallWithTextPB(svcAndMethod[1], ctx, requestPB)
	if err != nil {
		logger.Errorf("Error: %s", err)
	}
	if ctx.Metadata != nil {
		logger.Infof("Response metadata:\n%s", proto.MarshalTextString(ctx.Metadata))
	}
	if responsePB == nil {
		logger.Info("Response: <nil>")
	} else {
		logger.Infof("Response:\n%s", *responsePB)
	}
}

// DisplayStream creates a gRPC connection to the query target and makes a
// Subscribe call for the queried paths and streams the response via
// cfg.Display.
func DisplayStream(ctx context.Context, query Query, cfg *Config) error {
	c, err := createClient(ctx, query, cfg)
	if err != nil {
		return err
	}
	request, err := createSubscribeRequest(query)
	if err != nil {
		return err
	}
	stream, err := c.Subscribe(ctx)
	if err != nil {
		return err
	}
	if err := stream.Send(request); err != nil {
		return err
	}
	log.Infof("Subscribed with:\n%s", proto.MarshalTextString(request))
	for {
		resp, err := stream.Recv()
		log.V(2).Info(proto.MarshalTextString(resp))
		if err != nil {
			// TODO(hines): This should be io.EOF but for some reason the server
			// currently sends this code.
			if grpc.Code(err) != codes.OutOfRange {
				return err
			}
			return nil
		}
		switch resp.Response.(type) {
		default:
			log.Infof("Unknown response:\n%s\n", resp.String())
		case *ocpb.SubscribeResponse_Heartbeat:
			log.Infof("Heartbeat:%s\n", resp.String())
		case *ocpb.SubscribeResponse_Update:
			cfg.Display([]byte(proto.MarshalTextString(resp)))
		case *ocpb.SubscribeResponse_SyncResponse:
			log.Infof("Sync Response: %s", resp.String())
			if cfg.Once {
				stream.CloseSend()
				return nil
			}
		}
	}
}

// Save writes all domain configuration and policy data.
func (d *Domain) Save() error {
	file, err := util.CreatePath(d.ConfigPath, 0777, 0666)
	if err != nil {
		return err
	}
	ds := proto.MarshalTextString(&d.Config)
	fmt.Fprint(file, ds)
	file.Close()
	return d.Guard.Save(d.Keys.SigningKey)
}

func TestUnmarshaling(t *testing.T) {
	for _, tt := range unmarshalingTests {
		// Make a new instance of the type of our expected object.
		p := reflect.New(reflect.TypeOf(tt.pb).Elem()).Interface().(proto.Message)

		err := UnmarshalString(tt.json, p)
		if err != nil {
			t.Error(err)
			continue
		}

		// For easier diffs, compare text strings of the protos.
		exp := proto.MarshalTextString(tt.pb)
		act := proto.MarshalTextString(p)
		if string(exp) != string(act) {
			t.Errorf("%s: got [%s] want [%s]", tt.desc, act, exp)
		}
	}
}

func TestResolve(t *testing.T) { // are you function enough not to back down?
	// Test resolution on a simple two-package system:
	//
	// Package foo is compiled as test data from the source
	//    package foo
	//    func Foo() int { return 0 }
	//
	// Package bar is specified as source and imports foo.
	// TODO(fromberger): Compile foo as part of the build.
	foo, err := readTestFile("kythe/go/indexer/testdata/foo.a")
	if err != nil {
		t.Fatalf("Unable to read foo.a: %v", err)
	}
	const bar = `package bar

import "test/foo"

func init() { println(foo.Foo()) }
`
	fetcher := memFetcher{
		hexDigest(foo):         string(foo),
		hexDigest([]byte(bar)): bar,
	}
	unit := &apb.CompilationUnit{
		VName: &spb.VName{Language: "go", Corpus: "test", Path: "bar", Signature: ":pkg:"},
		RequiredInput: []*apb.CompilationUnit_FileInput{{
			VName: &spb.VName{Language: "go", Corpus: "test", Path: "foo", Signature: ":pkg:"},
			Info:  &apb.FileInfo{Path: "testdata/foo.a", Digest: hexDigest(foo)},
		}, {
			Info: &apb.FileInfo{Path: "testdata/bar.go", Digest: hexDigest([]byte(bar))},
		}},
		SourceFile: []string{"testdata/bar.go"},
	}
	pi, err := Resolve(unit, fetcher, nil)
	if err != nil {
		t.Fatalf("Resolve failed: %v\nInput unit:\n%s", err, proto.MarshalTextString(unit))
	}
	if got, want := pi.Name, "bar"; got != want {
		t.Errorf("Package name: got %q, want %q", got, want)
	}
	if got, want := pi.ImportPath, "test/bar"; got != want {
		t.Errorf("Import path: got %q, want %q", got, want)
	}
	if dep, ok := pi.Dependencies["test/foo"]; !ok {
		t.Errorf("Missing dependency for test/foo in %+v", pi.Dependencies)
	} else if pi.VNames[dep] == nil {
		t.Errorf("Missing VName for test/foo in %+v", pi.VNames)
	}
	if got, want := len(pi.Files), len(unit.SourceFile); got != want {
		t.Errorf("Source files: got %d, want %d", got, want)
	}
	for _, err := range pi.Errors {
		t.Errorf("Unexpected resolution error: %v", err)
	}
}

func main() {
	var params cmdParams
	flag.Var(&params, "param", "Task Parameter of the form key=value. May be repeated.")
	flag.Parse()

	if *task == "" {
		fmt.Println("Please specify a task using the --task parameter.")
		os.Exit(1)
	}
	if *automationServer == "" {
		fmt.Println("Please specify the automation server address using the --server parameter.")
		os.Exit(2)
	}

	fmt.Println("Connecting to Automation Server:", *automationServer)

	conn, err := grpc.Dial(*automationServer)
	if err != nil {
		fmt.Println("Cannot create connection:", err)
		os.Exit(3)
	}
	defer conn.Close()
	client := pbs.NewAutomationClient(conn)

	enqueueRequest := &pb.EnqueueClusterOperationRequest{
		Name:       *task,
		Parameters: params.parameters,
	}
	fmt.Printf("Sending request:\n%v", proto.MarshalTextString(enqueueRequest))
	enqueueResponse, err := client.EnqueueClusterOperation(context.Background(), enqueueRequest)
	if err != nil {
		fmt.Println("Failed to enqueue ClusterOperation. Error:", err)
		os.Exit(4)
	}
	fmt.Println("Operation was enqueued. Details:", enqueueResponse)
	resp, errWait := waitForClusterOp(client, enqueueResponse.Id)
	if errWait != nil {
		fmt.Println("ERROR:", errWait)
		os.Exit(5)
	}
	fmt.Printf("SUCCESS: ClusterOperation finished.\n\nDetails:\n%v", proto.MarshalTextString(resp))
}

func TestDecodeLinkTicketBasic(t *testing.T) {
	ticket := &replication.ServiceLinkTicket{
		PrimaryId:   proto.String("dummyid"),
		PrimaryUrl:  proto.String("http://dummy"),
		GeneratedBy: proto.String("dummy"),
		Ticket:      []byte("dummyticket"),
	}
	m, err := proto.Marshal(ticket)
	if err != nil {
		t.Fatalf("proto.Marshal()=%q,_; want <nil>", err)
	}
	str := encodeAndTrimEqual(m)
	dec, err := replication.DecodeLinkTicket(str)
	if err != nil {
		t.Errorf("DecodeLinkTicket(%q) = _, %q; want <nil>", str, err)
	}
	if !reflect.DeepEqual(ticket, dec) {
		t.Errorf("DecodeLinkTicket(%q) = %q; want %q", str, proto.MarshalTextString(dec), proto.MarshalTextString(ticket))
	}
}

func TestUnmarshalling(t *testing.T) {
	for _, tt := range unmarshallingTests {
		// Make a new instance of the type of our expected object.
		p := proto.Clone(tt.pb)
		p.Reset()

		err := UnmarshalString(tt.json, p)
		if err != nil {
			t.Error(err)
			continue
		}

		// For easier diffs, compare text strings of the protos.
		exp := proto.MarshalTextString(tt.pb)
		act := proto.MarshalTextString(p)
		if string(exp) != string(act) {
			t.Errorf("%s: got [%s] want [%s]", tt.desc, act, exp)
		}
	}
}

func TestParseTextFiles(t *testing.T) {
	tests := []struct {
		path string
		out  string
	}{
		{
			path: "fixtures/textfile/no_metric_files",
			out:  "fixtures/textfile/no_metric_files.out",
		},
		{
			path: "fixtures/textfile/two_metric_files",
			out:  "fixtures/textfile/two_metric_files.out",
		},
		{
			path: "fixtures/textfile/nonexistent_path",
			out:  "fixtures/textfile/nonexistent_path.out",
		},
	}

	for i, test := range tests {
		c := textFileCollector{
			path: test.path,
		}

		// Suppress a log message about `nonexistent_path` not existing, this is
		// expected and clutters the test output.
		err := flag.Set("log.level", "fatal")
		if err != nil {
			t.Fatal(err)
		}

		mfs := c.parseTextFiles()
		textMFs := make([]string, 0, len(mfs))
		for _, mf := range mfs {
			if mf.GetName() == "node_textfile_mtime" {
				mf.GetMetric()[0].GetGauge().Value = proto.Float64(1)
				mf.GetMetric()[1].GetGauge().Value = proto.Float64(2)
			}
			textMFs = append(textMFs, proto.MarshalTextString(mf))
		}
		sort.Strings(textMFs)
		got := strings.Join(textMFs, "")

		want, err := ioutil.ReadFile(test.out)
		if err != nil {
			t.Fatalf("%d. error reading fixture file %s: %s", i, test.out, err)
		}

		if string(want) != got {
			t.Fatalf("%d. want:\n\n%s\n\ngot:\n\n%s", i, string(want), got)
		}
	}
}