func TestGetBlockFromPeerAfterPeerAnnounces(t *testing.T) {

	net := tn.VirtualNetwork(mockrouting.NewServer(), delay.Fixed(kNetworkDelay))
	block := blocks.NewBlock([]byte("block"))
	g := NewTestSessionGenerator(net)
	defer g.Close()

	peers := g.Instances(2)
	hasBlock := peers[0]
	defer hasBlock.Exchange.Close()

	if err := hasBlock.Exchange.HasBlock(block); err != nil {
		t.Fatal(err)
	}

	wantsBlock := peers[1]
	defer wantsBlock.Exchange.Close()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	received, err := wantsBlock.Exchange.GetBlock(ctx, block.Cid())
	if err != nil {
		t.Log(err)
		t.Fatal("Expected to succeed")
	}

	if !bytes.Equal(block.RawData(), received.RawData()) {
		t.Fatal("Data doesn't match")
	}
}

func TestChain(t *testing.T) {
	mockCtrl := gomock.NewController(t)
	defer mockCtrl.Finish()

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	req := &transport.Request{
		Caller:    "somecaller",
		Service:   "someservice",
		Encoding:  transport.Encoding("raw"),
		Procedure: "hello",
		Body:      bytes.NewReader([]byte{1, 2, 3}),
	}
	res := &transport.Response{
		Body: ioutil.NopCloser(bytes.NewReader([]byte{4, 5, 6})),
	}

	o := transporttest.NewMockUnaryOutbound(mockCtrl)
	o.EXPECT().Call(ctx, req).After(
		o.EXPECT().Call(ctx, req).Return(nil, errors.New("great sadness")),
	).Return(res, nil)

	before := &countFilter{}
	after := &countFilter{}
	gotRes, err := transport.ApplyFilter(
		o, Chain(before, retryFilter, after)).Call(ctx, req)

	assert.NoError(t, err, "expected success")
	assert.Equal(t, 1, before.Count, "expected outer filter to be called once")
	assert.Equal(t, 2, after.Count, "expected inner filter to be called twice")
	assert.Equal(t, res, gotRes, "expected response to match")
}

func TestBasicBitswap(t *testing.T) {
	net := tn.VirtualNetwork(mockrouting.NewServer(), delay.Fixed(kNetworkDelay))
	sg := NewTestSessionGenerator(net)
	defer sg.Close()
	bg := blocksutil.NewBlockGenerator()

	t.Log("Test a one node trying to get one block from another")

	instances := sg.Instances(2)
	blocks := bg.Blocks(1)
	err := instances[0].Exchange.HasBlock(blocks[0])
	if err != nil {
		t.Fatal(err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
	defer cancel()
	blk, err := instances[1].Exchange.GetBlock(ctx, blocks[0].Cid())
	if err != nil {
		t.Fatal(err)
	}

	t.Log(blk)
	for _, inst := range instances {
		err := inst.Exchange.Close()
		if err != nil {
			t.Fatal(err)
		}
	}
}

func runReg(args []string) {
	key, err := anyKey(filepath.Join(configDir, accountKey), regGen)
	if err != nil {
		fatalf("account key: %v", err)
	}
	uc := &userConfig{
		Account: acme.Account{Contact: args},
		key:     key,
	}

	prompt := ttyPrompt
	if regAccept {
		prompt = acme.AcceptTOS
	}
	client := &acme.Client{
		Key:          uc.key,
		DirectoryURL: string(regDisco),
	}

	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	a, err := client.Register(ctx, &uc.Account, prompt)
	if err != nil {
		fatalf("%v", err)
	}
	uc.Account = *a
	if err := writeConfig(uc); err != nil {
		errorf("write config: %v", err)
	}
}

// GetDiagnostic runs a diagnostics request across the entire network
func (d *Diagnostics) GetDiagnostic(ctx context.Context, timeout time.Duration) ([]*DiagInfo, error) {
	log.Debug("getting diagnostic")
	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	diagID := newID()
	d.diagLock.Lock()
	d.diagMap[diagID] = time.Now()
	d.diagLock.Unlock()

	log.Debug("begin diagnostic")

	peers := d.getPeers()
	log.Debugf("Sending diagnostic request to %d peers.", len(peers))

	pmes := newMessage(diagID)

	pmes.SetTimeoutDuration(timeout - HopTimeoutDecrement) // decrease timeout per hop
	dpeers, err := d.getDiagnosticFromPeers(ctx, d.getPeers(), pmes)
	if err != nil {
		return nil, fmt.Errorf("diagnostic from peers err: %s", err)
	}

	di := d.getDiagInfo()
	out := []*DiagInfo{di}
	for dpi := range dpeers {
		out = append(out, dpi)
	}
	return out, nil
}

func (c *containerBase) start(ctx context.Context) error {
	// make sure we have vm
	if c.vm == nil {
		return NotYetExistError{c.ExecConfig.ID}
	}

	// Power on
	_, err := c.vm.WaitForResult(ctx, func(ctx context.Context) (tasks.Task, error) {
		return c.vm.PowerOn(ctx)
	})
	if err != nil {
		return err
	}

	// guestinfo key that we want to wait for
	key := fmt.Sprintf("guestinfo.vice..sessions|%s.started", c.ExecConfig.ID)
	var detail string

	// Wait some before giving up...
	ctx, cancel := context.WithTimeout(ctx, propertyCollectorTimeout)
	defer cancel()

	detail, err = c.vm.WaitForKeyInExtraConfig(ctx, key)
	if err != nil {
		return fmt.Errorf("unable to wait for process launch status: %s", err.Error())
	}

	if detail != "true" {
		return errors.New(detail)
	}

	return nil
}

// Query sends a command to the server and returns the Response
func Query(c clientset.Interface, query string) (*influxdb.Response, error) {

	ctx, cancel := context.WithTimeout(context.Background(), framework.SingleCallTimeout)
	defer cancel()

	result, err := c.Core().RESTClient().Get().
		Prefix("proxy").
		Namespace("kube-system").
		Resource("services").
		Name(influxdbService+":api").
		Suffix("query").
		Param("q", query).
		Param("db", influxdbDatabaseName).
		Param("epoch", "s").
		Do().
		Raw()

	if err != nil {
		if ctx.Err() != nil {
			framework.Failf("Failed to query influx db: %v", err)
		}
		return nil, err
	}

	var response influxdb.Response
	dec := json.NewDecoder(bytes.NewReader(result))
	dec.UseNumber()
	err = dec.Decode(&response)

	if err != nil {
		return nil, err
	}
	return &response, nil
}

func (s *Supervised) call(f func(Client) error) error {
	c, err := s.dcf()
	if err != nil {
		return err
	}

	ctx := c.Context()
	if err = f(c); err != ErrDisconnected {
		return err
	}

	// Wait for new client.
	ctx, cancel := context.WithTimeout(ctx, s.timeout)
	defer cancel()

	if <-ctx.Done(); ctx.Err() == context.DeadlineExceeded {
		// Client is still disconnected. Return it as is.
		return ErrDisconnected
	}

	// Previous context was canceled. This means that the client changed.
	c, err = s.dcf()
	if err != nil {
		return err
	}

	return f(c)
}

func TestRecording(t *testing.T) {
	tMock := testingTMock{t, 0}

	dir, err := ioutil.TempDir("", "yarpcgorecorder")
	if err != nil {
		t.Fatal(err)
	}
	defer os.RemoveAll(dir) // clean up

	recorder := NewRecorder(&tMock, RecordMode(Append), RecordsPath(dir))

	withConnectedClient(t, recorder, func(client raw.Client) {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()

		rbody, _, err := client.Call(ctx, yarpc.NewReqMeta().Procedure("hello"), []byte("Hello"))
		require.NoError(t, err)
		assert.Equal(t, []byte("Hello, World"), rbody)
	})

	recordPath := path.Join(dir, refRecordFilename)
	_, err = os.Stat(recordPath)
	require.NoError(t, err)

	recordContent, err := ioutil.ReadFile(recordPath)
	require.NoError(t, err)
	assert.Equal(t, refRecordContent, string(recordContent))
}

func runUpdate(args []string) {
	uc, err := readConfig()
	if err != nil {
		fatalf("read config: %v", err)
	}
	if uc.key == nil {
		fatalf("no key found for %s", uc.URI)
	}

	client := acme.Client{Key: uc.key}
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	if updateAccept {
		a, err := client.GetReg(ctx, uc.URI)
		if err != nil {
			fatalf(err.Error())
		}
		uc.Account = *a
		uc.AgreedTerms = a.CurrentTerms
	}
	if len(args) != 0 {
		uc.Contact = args
	}

	a, err := client.UpdateReg(ctx, &uc.Account)
	if err != nil {
		fatalf(err.Error())
	}
	uc.Account = *a
	if err := writeConfig(uc); err != nil {
		fatalf("write config: %v", err)
	}
	printAccount(os.Stdout, &uc.Account, filepath.Join(configDir, accountKey))
}

// ResolveLinks iteratively resolves names by walking the link hierarchy.
// Every node is fetched from the DAGService, resolving the next name.
// Returns the list of nodes forming the path, starting with ndd. This list is
// guaranteed never to be empty.
//
// ResolveLinks(nd, []string{"foo", "bar", "baz"})
// would retrieve "baz" in ("bar" in ("foo" in nd.Links).Links).Links
func (s *Resolver) ResolveLinks(ctx context.Context, ndd node.Node, names []string) ([]node.Node, error) {

	result := make([]node.Node, 0, len(names)+1)
	result = append(result, ndd)
	nd := ndd // dup arg workaround

	// for each of the path components
	for len(names) > 0 {
		var cancel context.CancelFunc
		ctx, cancel = context.WithTimeout(ctx, time.Minute)
		defer cancel()

		lnk, rest, err := nd.ResolveLink(names)
		if err == dag.ErrLinkNotFound {
			return result, ErrNoLink{Name: names[0], Node: nd.Cid()}
		} else if err != nil {
			return result, err
		}

		nextnode, err := lnk.GetNode(ctx, s.DAG)
		if err != nil {
			return result, err
		}

		nd = nextnode
		result = append(result, nextnode)
		names = rest
	}
	return result, nil
}

// handleError makes the request to the StackDriver Error Reporting API
func handleError(errorsClient *errors.Client, d device) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	log.Printf("Sending report for %s (%s)", d.Name, d.Id)
	errorsClient.Reportf(ctx, nil, "Device is offline: %s (%s)", d.Name, d.Id)
}

// ParseTTL takes a context parses the given TTL, clamping the context to that TTL
// and as a side-effect, tracking any errors encountered while attempting to
// parse and validate that TTL. Should only be used for unary requests
func (v *Validator) ParseTTL(ctx context.Context, ttl string) (context.Context, func()) {
	if ttl == "" {
		// The TTL is missing so set it to 0 and let Validate() fail with the
		// correct error message.
		return ctx, func() {}
	}

	ttlms, err := strconv.Atoi(ttl)
	if err != nil {
		v.errTTL = invalidTTLError{
			Service:   v.Request.Service,
			Procedure: v.Request.Procedure,
			TTL:       ttl,
		}
		return ctx, func() {}
	}
	// negative TTLs are invalid
	if ttlms < 0 {
		v.errTTL = invalidTTLError{
			Service:   v.Request.Service,
			Procedure: v.Request.Procedure,
			TTL:       fmt.Sprint(ttlms),
		}
		return ctx, func() {}
	}

	return context.WithTimeout(ctx, time.Duration(ttlms)*time.Millisecond)
}

func main() {
	// Pass a context with a timeout to tell a blocking function that it
	// should abandon its work after the timeout elapses.

	ctx, cancel := context.WithTimeout(context.Background(), 3000*time.Millisecond)

	ctx = context.WithValue(ctx, "a", 42)

	go t(ctx)
	// time.Sleep(time.Millisecond * 200)
	// cancel()
	select {
	case <-time.After(4 * time.Second):
		fmt.Println("overslept")
	case <-ctx.Done():
		fmt.Println(ctx.Err()) // prints "context deadline exceeded"
	}

	// Even though ctx should have expired already, it is good
	// practice to call its cancelation function in any case.
	// Failure to do so may keep the context and its parent alive
	// longer than necessary.
	fmt.Println(ctx.Value("a"))
	cancel()

}

func TestNopFilter(t *testing.T) {
	mockCtrl := gomock.NewController(t)
	defer mockCtrl.Finish()

	o := transporttest.NewMockUnaryOutbound(mockCtrl)
	wrappedO := transport.ApplyFilter(o, transport.NopFilter)

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	req := &transport.Request{
		Caller:    "somecaller",
		Service:   "someservice",
		Encoding:  raw.Encoding,
		Procedure: "hello",
		Body:      bytes.NewReader([]byte{1, 2, 3}),
	}

	res := &transport.Response{Body: ioutil.NopCloser(bytes.NewReader([]byte{4, 5, 6}))}
	o.EXPECT().Call(ctx, req).Return(res, nil)

	got, err := wrappedO.Call(ctx, req)
	if assert.NoError(t, err) {
		assert.Equal(t, res, got)
	}
}

func main() {
	configFile := flag.String("config-file", "", "Config file for RETS connection")
	metadataFile := flag.String("metadata-options", "", "Config file for metadata options")
	output := flag.String("output", "", "Directory for file output")

	config := common.Config{}
	config.SetFlags()

	metadataOpts := MetadataOptions{}
	metadataOpts.SetFlags()

	flag.Parse()

	if *configFile != "" {
		err := config.LoadFrom(*configFile)
		if err != nil {
			panic(err)
		}
	}
	fmt.Printf("Connection Settings: %v\n", config)
	if *metadataFile != "" {
		err := metadataOpts.LoadFrom(*metadataFile)
		if err != nil {
			panic(err)
		}
	}
	fmt.Printf("Search Options: %v\n", metadataOpts)

	// should we throw an err here too?
	session, err := config.Initialize()
	if err != nil {
		panic(err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
	defer cancel()

	capability, err := rets.Login(session, ctx, rets.LoginRequest{URL: config.URL})
	if err != nil {
		panic(err)
	}
	defer rets.Logout(session, ctx, rets.LogoutRequest{URL: capability.Logout})

	reader, err := rets.MetadataStream(session, ctx, rets.MetadataRequest{
		URL:    capability.GetMetadata,
		Format: metadataOpts.Format,
		MType:  metadataOpts.MType,
		ID:     metadataOpts.ID,
	})
	defer reader.Close()
	if err != nil {
		panic(err)
	}
	out := os.Stdout
	if *output != "" {
		out, _ = os.Create(*output + "/metadata.xml")
		defer out.Close()
	}
	io.Copy(out, reader)
}

// pipeStream relays over a stream to a remote peer. It's like `cat`
func (rs *RelayService) pipeStream(src, dst peer.ID, s inet.Stream) error {
	// TODO: find a good way to pass contexts into here
	nsctx, cancel := context.WithTimeout(context.TODO(), time.Second*30)
	defer cancel()

	s2, err := rs.openStreamToPeer(nsctx, dst)
	if err != nil {
		return fmt.Errorf("failed to open stream to peer: %s -- %s", dst, err)
	}
	cancel() // cancel here because this function might last a while

	if err := WriteHeader(s2, src, dst); err != nil {
		return err
	}

	// connect the series of tubes.
	done := make(chan retio, 2)
	go func() {
		n, err := io.Copy(s2, s)
		done <- retio{n, err}
	}()
	go func() {
		n, err := io.Copy(s, s2)
		done <- retio{n, err}
	}()

	r1 := <-done
	r2 := <-done
	log.Infof("%s relayed %d/%d bytes between %s and %s", rs.host.ID(), r1.n, r2.n, src, dst)

	if r1.err != nil {
		return r1.err
	}
	return r2.err
}

func TestPutManyAddsToBloom(t *testing.T) {
	bs := NewBlockstore(syncds.MutexWrap(ds.NewMapDatastore()))

	ctx, _ := context.WithTimeout(context.Background(), 1*time.Second)
	cachedbs, err := testBloomCached(bs, ctx)

	select {
	case <-cachedbs.rebuildChan:
	case <-ctx.Done():
		t.Fatalf("Timeout wating for rebuild: %d", cachedbs.bloom.ElementsAdded())
	}

	block1 := blocks.NewBlock([]byte("foo"))
	block2 := blocks.NewBlock([]byte("bar"))

	cachedbs.PutMany([]blocks.Block{block1})
	has, err := cachedbs.Has(block1.Cid())
	if err != nil {
		t.Fatal(err)
	}
	if has == false {
		t.Fatal("added block is reported missing")
	}

	has, err = cachedbs.Has(block2.Cid())
	if err != nil {
		t.Fatal(err)
	}
	if has == true {
		t.Fatal("not added block is reported to be in blockstore")
	}
}

func (client *Client) ping(ws *websocket.Conn, addr string) {
	log.WithField("port", client.Port).Infoln("DailTLS ok: " + addr)
	info, err := client.getServerInfo()
	if err != nil {
		log.WithField("port", client.Port).Errorln("getServerInfo", err)
		return
	}

	ticker := time.NewTicker(info.PingSecond * time.Second)
	defer func() {
		ticker.Stop()
		ws.Close()
		log.WithField("port", client.Port).Infoln("Ws closed")
	}()
	log.WithField("port", client.Port).Infoln("Ws started")

	req := client.innerRequest("HEAD", HOST_OK)

	for {
		select {
		case <-ticker.C:
			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
			res, err := client.h2Transport.RoundTrip(req.WithContext(ctx))
			if err != nil || res.StatusCode != http.StatusOK {
				cancel()
				return
			}
			cancel()
		}
	}
}

// readTransactions reads # of transactions from the k8petstore web server endpoint.
// for more details see the source of the k8petstore web server.
func readTransactions(c clientset.Interface, ns string) (error, int) {
	proxyRequest, errProxy := framework.GetServicesProxyRequest(c, c.Core().RESTClient().Get())
	if errProxy != nil {
		return errProxy, -1
	}

	ctx, cancel := context.WithTimeout(context.Background(), framework.SingleCallTimeout)
	defer cancel()

	body, err := proxyRequest.Namespace(ns).
		Context(ctx).
		Name("frontend").
		Suffix("llen").
		DoRaw()
	if err != nil {
		if ctx.Err() != nil {
			framework.Failf("Failed to read petstore transactions: %v", err)
		}
		return err, -1
	}

	totalTrans, err := strconv.Atoi(string(body))
	return err, totalTrans

}