func Example_serviceAccount() {
	// Warning: The better way to use service accounts is to set GOOGLE_APPLICATION_CREDENTIALS
	// and use the Application Default Credentials.
	ctx := context.Background()
	// Use a JSON key file associated with a Google service account to
	// authenticate and authorize.
	// Go to https://console.developers.google.com/permissions/serviceaccounts to create
	// and download a service account key for your project.
	//
	// Note: The example uses the datastore client, but the same steps apply to
	// the other client libraries underneath this package.
	key, err := ioutil.ReadFile("/path/to/service-account-key.json")
	if err != nil {
		// TODO: handle error.
	}
	cfg, err := google.JWTConfigFromJSON(key, datastore.ScopeDatastore)
	if err != nil {
		// TODO: handle error.
	}
	client, err := datastore.NewClient(
		ctx, "project-id", option.WithTokenSource(cfg.TokenSource(ctx)))
	if err != nil {
		// TODO: handle error.
	}
	// Use the client.
	_ = client
}

func (s *shard) createOutputFile(c context.Context) (io.WriteCloser, error) {
	c, _ = context.WithTimeout(c, time.Duration(10)*time.Minute)
	// for development we can't use the appengine default credentials so
	// instead need to create our own oauth token source to access storage

	// TODO: maybe give job a chance to generate this - it could also
	// create the writer (?). The only reason we're doing it is to prevent
	// duplication and also handle the file rollup operations
	var client *cstorage.Client
	if appengine.IsDevAppServer() {
		jsonKey, err := ioutil.ReadFile("service-account.json")
		if err != nil {
			return nil, err
		}
		conf, err := google.JWTConfigFromJSON(jsonKey, cstorage.ScopeReadWrite)
		if err != nil {
			return nil, err
		}
		client, err = cstorage.NewClient(c, option.WithTokenSource(conf.TokenSource(c)))
		if err != nil {
			return nil, err
		}
	} else {
		var err error
		client, err = cstorage.NewClient(c)
		if err != nil {
			return nil, err
		}
	}

	o := client.Bucket(s.job.Bucket).Object(s.sliceFilename(s.Sequence)).NewWriter(c)

	// TODO: wrap writer to count bytes and continue slice if we get close to 10Mb limit (?)
	return o, nil
}

func TestIntegrationPingBadProject(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}

	ctx := context.Background()
	ts := testutil.TokenSource(ctx, Scope)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	for _, projID := range []string{
		testutil.ProjID() + "-BAD", // nonexistent project
		"amazing-height-519",       // exists, but wrong creds
	} {
		c, err := NewClient(ctx, projID, "logging-integration-test", option.WithTokenSource(ts))
		if err != nil {
			t.Fatalf("project %s: error creating client: %v", projID, err)
		}
		if err := c.Ping(); err == nil {
			t.Errorf("project %s: want error pinging logging api, got nil", projID)
		}
		// Ping twice, just to make sure the deduping doesn't mess with the result.
		if err := c.Ping(); err == nil {
			t.Errorf("project %s: want error pinging logging api, got nil", projID)
		}
	}
}

func NewGCS(name string, info map[string]string) (Backend, error) {
	b := &gcsBackend{
		name:       name,
		bucketName: info["bucket"],
	}
	keyJSON := []byte(info["key"])

	if b.bucketName == "" {
		return nil, fmt.Errorf("blobstore: missing Google Cloud Storage bucket param for %s", name)
	}
	if len(keyJSON) == 0 {
		return nil, fmt.Errorf("blobstore: missing Google Cloud Storage key JSON param for %s", name)
	}

	jwtToken, err := google.JWTConfigFromJSON(keyJSON, "https://www.googleapis.com/auth/devstorage.read_write")
	if err != nil {
		return nil, fmt.Errorf("blobstore: error loading Google Cloud Storage JSON key: %s", err)
	}
	tokenSource := jwtToken.TokenSource(context.Background())

	// Test getting an OAuth token so we can disambiguate an issue with the
	// token and an issue with the bucket permissions below.
	if _, err := tokenSource.Token(); err != nil {
		return nil, fmt.Errorf("blobstore: error getting Google Cloud Storage OAuth token: %s", err)
	}

	pemBlock, _ := pem.Decode(jwtToken.PrivateKey)
	privateKey, err := x509.ParsePKCS8PrivateKey(pemBlock.Bytes)
	if err != nil {
		return nil, fmt.Errorf("blobstore: error decoding Google Cloud Storage private key: %s", err)
	}
	rsaPrivateKey, ok := privateKey.(*rsa.PrivateKey)
	if !ok {
		return nil, fmt.Errorf("blobstore: unexpected Google Cloud Storage key type: %T", privateKey)
	}
	b.signOpts = func() *storage.SignedURLOptions {
		return &storage.SignedURLOptions{
			GoogleAccessID: jwtToken.Email,
			SignBytes: func(b []byte) ([]byte, error) {
				digest := sha256.Sum256(b)
				return rsa.SignPKCS1v15(rand.Reader, rsaPrivateKey, crypto.SHA256, digest[:])
			},
			Method:  "GET",
			Expires: time.Now().Add(10 * time.Minute),
		}
	}

	client, err := storage.NewClient(context.Background(), option.WithTokenSource(tokenSource))
	if err != nil {
		return nil, fmt.Errorf("blobstore: error creating Google Cloud Storage client: %s", err)
	}
	b.bucket = client.Bucket(b.bucketName)

	_, err = b.bucket.Attrs(context.Background())
	if err != nil {
		return nil, fmt.Errorf("blobstore: error checking Google Cloud Storage bucket %q existence, ensure that it exists and Owner access for %s is included the bucket ACL: %q", b.bucketName, jwtToken.Email, err)
	}
	return b, nil
}

func upload(srcPath string) {
	if !*flagUpload {
		return
	}
	destName := strings.Replace(filepath.Base(srcPath), "camlistore", "camlistore-"+releaseDate.Format(fileDateFormat), 1)
	versionedTarball := "monthly/" + destName

	log.Printf("Uploading %s/%s ...", bucket, versionedTarball)

	ts, err := tokenSource(bucket)
	if err != nil {
		log.Fatal(err)
	}
	ctx := context.Background()
	stoClient, err := storage.NewClient(ctx, option.WithTokenSource(ts), option.WithHTTPClient(oauth2.NewClient(ctx, ts)))
	if err != nil {
		log.Fatal(err)
	}
	w := stoClient.Bucket(bucket).Object(versionedTarball).NewWriter(ctx)
	w.ACL = publicACL(project)
	w.CacheControl = "no-cache" // TODO: remove for non-tip releases? set expirations?
	contentType := "application/x-gtar"
	if strings.HasSuffix(versionedTarball, ".zip") {
		contentType = "application/zip"
	}
	w.ContentType = contentType
	csw := sha256.New()
	mw := io.MultiWriter(w, csw)

	src, err := os.Open(srcPath)
	if err != nil {
		log.Fatal(err)
	}
	defer src.Close()

	if _, err := io.Copy(mw, src); err != nil {
		log.Fatalf("io.Copy: %v", err)
	}
	if err := w.Close(); err != nil {
		log.Fatalf("closing GCS storage writer: %v", err)
	}
	log.Printf("Uploaded monthly tarball to %s", versionedTarball)

	// And upload the corresponding checksum
	checkSumFile := versionedTarball + ".sha256"
	sum := fmt.Sprintf("%x", csw.Sum(nil))
	w = stoClient.Bucket(bucket).Object(checkSumFile).NewWriter(ctx)
	w.ACL = publicACL(project)
	w.CacheControl = "no-cache" // TODO: remove for non-tip releases? set expirations?
	w.ContentType = "text/plain"
	if _, err := io.Copy(w, strings.NewReader(sum)); err != nil {
		log.Fatalf("error uploading checksum %v: %v", checkSumFile, err)
	}
	if err := w.Close(); err != nil {
		log.Fatalf("closing GCS storage writer: %v", err)
	}
	log.Printf("Uploaded monthly tarball checksum to %s", checkSumFile)
}

func getCredentialOpts(opts []option.ClientOption) []option.ClientOption {
	if ts := config.TokenSource; ts != nil {
		opts = append(opts, option.WithTokenSource(ts))
	}
	if tlsCreds := config.TLSCreds; tlsCreds != nil {
		opts = append(opts, option.WithGRPCDialOption(grpc.WithTransportCredentials(tlsCreds)))
	}
	return opts
}

func newClient(ctx context.Context, t *testing.T) *Client {
	ts := testutil.TokenSource(ctx, ScopeDatastore)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}
	client, err := NewClient(ctx, testutil.ProjID(), option.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("NewClient: %v", err)
	}
	return client
}

// uploadReleaseTarball uploads the generated tarball of binaries in
// camlistore-release/VERSION/camlistoreVERSION-REV-CONTENTS.EXT. It then makes a copy in
// the same bucket and path, as camlistoreVERSION-CONTENTS.EXT.
func uploadReleaseTarball() {
	proj := "camlistore-website"
	bucket := "camlistore-release"
	tarball := *flagVersion + "/" + filepath.Base(releaseTarball)
	versionedTarball := strings.Replace(tarball, "camlistore"+*flagVersion, "camlistore"+*flagVersion+"-"+rev(), 1)

	log.Printf("Uploading %s/%s ...", bucket, versionedTarball)

	ts, err := tokenSource(bucket)
	if err != nil {
		log.Fatal(err)
	}
	ctx := context.Background()
	stoClient, err := storage.NewClient(ctx, option.WithTokenSource(ts), option.WithHTTPClient(oauth2.NewClient(ctx, ts)))
	if err != nil {
		log.Fatal(err)
	}
	w := stoClient.Bucket(bucket).Object(versionedTarball).NewWriter(ctx)
	w.ACL = publicACL(proj)
	w.CacheControl = "no-cache" // TODO: remove for non-tip releases? set expirations?
	contentType := "application/x-gtar"
	if *buildOS == "windows" {
		contentType = "application/zip"
	}
	w.ContentType = contentType

	src, err := os.Open(releaseTarball)
	if err != nil {
		log.Fatal(err)
	}
	defer src.Close()

	if _, err := io.Copy(w, src); err != nil {
		log.Fatalf("io.Copy: %v", err)
	}
	if err := w.Close(); err != nil {
		log.Fatalf("closing GCS storage writer: %v", err)
	}
	log.Printf("Uploaded tarball to %s", versionedTarball)
	if !isWIP() {
		log.Printf("Copying tarball to %s/%s ...", bucket, tarball)
		dest := stoClient.Bucket(bucket).Object(tarball)
		if _, err := stoClient.Bucket(bucket).Object(versionedTarball).CopyTo(
			ctx,
			dest,
			&storage.ObjectAttrs{
				ACL:         publicACL(proj),
				ContentType: contentType,
			}); err != nil {
			log.Fatalf("Error uploading %v: %v", tarball, err)
		}
		log.Printf("Uploaded tarball to %s", tarball)
	}
}

func authOption() option.ClientOption {
	ts := testutil.TokenSource(context.Background(), Scope)
	if ts != nil {
		log.Println("authenticating via OAuth2")
		return option.WithTokenSource(ts)
	}
	apiKey := os.Getenv("GCLOUD_TESTS_API_KEY")
	if apiKey != "" {
		log.Println("authenticating with API key")
		return option.WithAPIKey(apiKey)
	}
	return nil
}

func integrationTestClient(ctx context.Context, t *testing.T) *Client {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}
	ts := testutil.TokenSource(ctx, Scope)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}
	client, err := NewClient(ctx, option.WithTokenSource(ts))
	if err != nil {
		t.Fatal(err)
	}
	return client
}

// config is like testConfig, but it doesn't need a *testing.T.
func config(ctx context.Context) (*Client, string) {
	ts := testutil.TokenSource(ctx, ScopeFullControl)
	if ts == nil {
		return nil, ""
	}
	p := testutil.ProjID()
	if p == "" {
		log.Fatal("The project ID must be set. See CONTRIBUTING.md for details")
	}
	client, err := NewClient(ctx, option.WithTokenSource(ts))
	if err != nil {
		log.Fatalf("NewClient: %v", err)
	}
	return client, p + suffix
}

func TestIntegration(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}

	ctx := context.Background()
	ts := testutil.TokenSource(ctx, Scope)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	projID := testutil.ProjID()

	c, err := NewClient(ctx, projID, "logging-integration-test", option.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("error creating client: %v", err)
	}

	if err := c.Ping(); err != nil {
		t.Fatalf("error pinging logging api: %v", err)
	}
	// Ping twice, to verify that deduping doesn't change the result.
	if err := c.Ping(); err != nil {
		t.Fatalf("error pinging logging api: %v", err)
	}

	if err := c.LogSync(Entry{Payload: customJSONObject{}}); err != nil {
		t.Fatalf("error writing log: %v", err)
	}

	if err := c.Log(Entry{Payload: customJSONObject{}}); err != nil {
		t.Fatalf("error writing log: %v", err)
	}

	if _, err := c.Writer(Default).Write([]byte("test log with io.Writer")); err != nil {
		t.Fatalf("error writing log using io.Writer: %v", err)
	}

	c.Logger(Default).Println("test log with log.Logger")

	if err := c.Flush(); err != nil {
		t.Fatalf("error flushing logs: %v", err)
	}
}

func newLogTest(t *testing.T) *logTest {
	handlerc := make(chan http.Handler, 1)
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		select {
		case h := <-handlerc:
			h.ServeHTTP(w, r)
		default:
			slurp, _ := ioutil.ReadAll(r.Body)
			t.Errorf("Unexpected HTTP request received: %s", slurp)
			w.WriteHeader(500)
			io.WriteString(w, "unexpected HTTP request")
		}
	}))
	c, err := NewClient(context.Background(), "PROJ-ID", "LOG-NAME",
		option.WithEndpoint(ts.URL),
		option.WithTokenSource(dummyTokenSource{}), // prevent DefaultTokenSource
	)
	if err != nil {
		t.Fatal(err)
	}
	var clock struct {
		sync.Mutex
		now time.Time
	}
	c.timeNow = func() time.Time {
		clock.Lock()
		defer clock.Unlock()
		if clock.now.IsZero() {
			clock.now = time.Unix(0, 0)
		}
		clock.now = clock.now.Add(1 * time.Second)
		return clock.now
	}
	return &logTest{
		t:        t,
		ts:       ts,
		c:        c,
		handlerc: handlerc,
	}
}

// If integration tests will be run, create a unique bucket for them.
func initIntegrationTest() {
	flag.Parse() // needed for testing.Short()
	if testing.Short() {
		return
	}
	ctx := context.Background()
	ts := testutil.TokenSource(ctx, Scope)
	if ts == nil {
		log.Println("Integration tests skipped. See CONTRIBUTING.md for details")
		return
	}
	projID := testutil.ProjID()
	var err error
	client, err = NewClient(ctx, projID, option.WithTokenSource(ts))
	if err != nil {
		log.Fatalf("NewClient: %v", err)
	}
	dataset = client.Dataset("bigquery_integration_test")
	if err := dataset.Create(ctx); err != nil && !hasStatusCode(err, http.StatusConflict) { // AlreadyExists is 409
		log.Fatalf("creating dataset: %v", err)
	}
}

// Check that user optioned grpc.WithDialer option overrides the App Engine hook.
func TestGRPCHook(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
	expected := false

	appengineDialerHook = (func(ctx context.Context) grpc.DialOption {
		return grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
			t.Error("did not expect a call to notExpected dialer, got one")
			cancel()
			return nil, errors.New("not expected")
		})
	})
	defer func() {
		appengineDialerHook = nil
	}()

	expectedDialer := grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
		expected = true
		cancel()
		return nil, errors.New("expected")
	})

	conn, err := DialGRPC(ctx,
		option.WithTokenSource(oauth2.StaticTokenSource(nil)), // No creds.
		option.WithGRPCDialOption(expectedDialer),
		option.WithEndpoint("example.google.com:443"))
	if err != nil {
		t.Errorf("DialGRPC: error %v, want nil", err)
	}

	// gRPC doesn't connect before the first call.
	grpc.Invoke(ctx, "foo", nil, nil, conn)
	conn.Close()

	if !expected {
		t.Error("expected a call to expected dialer, didn't get one")
	}
}

// uploadDockerImage makes a tar.gz snapshot of the camlistored docker image,
// and uploads it at camlistore-release/docker/camlistored-REV.tar.gz. It then
// makes a copy in the same bucket and path as camlistored.tar.gz.
func uploadDockerImage() {
	proj := "camlistore-website"
	bucket := "camlistore-release"
	versionedTarball := "docker/camlistored-" + rev() + ".tar.gz"
	tarball := "docker/camlistored.tar.gz"
	versionFile := "docker/VERSION"

	log.Printf("Uploading %s/%s ...", bucket, versionedTarball)

	ts, err := tokenSource(bucket)
	if err != nil {
		log.Fatal(err)
	}
	ctx := context.Background()
	stoClient, err := storage.NewClient(ctx, option.WithTokenSource(ts), option.WithHTTPClient(oauth2.NewClient(ctx, ts)))
	if err != nil {
		log.Fatal(err)
	}
	w := stoClient.Bucket(bucket).Object(versionedTarball).NewWriter(ctx)
	w.ACL = publicACL(proj)
	w.CacheControl = "no-cache" // TODO: remove for non-tip releases? set expirations?
	w.ContentType = "application/x-gtar"

	dockerSave := exec.Command("docker", "save", serverImage)
	dockerSave.Stderr = os.Stderr
	tar, err := dockerSave.StdoutPipe()
	if err != nil {
		log.Fatal(err)
	}
	targz, pw := io.Pipe()
	go func() {
		zw := gzip.NewWriter(pw)
		n, err := io.Copy(zw, tar)
		if err != nil {
			log.Fatalf("Error copying to gzip writer: after %d bytes, %v", n, err)
		}
		if err := zw.Close(); err != nil {
			log.Fatalf("gzip.Close: %v", err)
		}
		pw.CloseWithError(err)
	}()
	if err := dockerSave.Start(); err != nil {
		log.Fatalf("Error starting docker save %v: %v", serverImage, err)
	}
	if _, err := io.Copy(w, targz); err != nil {
		log.Fatalf("io.Copy: %v", err)
	}
	if err := w.Close(); err != nil {
		log.Fatalf("closing GCS storage writer: %v", err)
	}
	if err := dockerSave.Wait(); err != nil {
		log.Fatalf("Error waiting for docker save %v: %v", serverImage, err)
	}
	log.Printf("Uploaded tarball to %s", versionedTarball)
	if isWIP() {
		return
	}
	log.Printf("Copying tarball to %s/%s ...", bucket, tarball)
	dest := stoClient.Bucket(bucket).Object(tarball)
	if _, err := stoClient.Bucket(bucket).Object(versionedTarball).CopyTo(
		ctx,
		dest,
		&storage.ObjectAttrs{
			ACL:          publicACL(proj),
			CacheControl: "no-cache",
			ContentType:  "application/x-gtar",
		}); err != nil {
		log.Fatalf("Error uploading %v: %v", tarball, err)
	}
	log.Printf("Uploaded tarball to %s", tarball)

	log.Printf("Updating %s/%s file...", bucket, versionFile)
	w = stoClient.Bucket(bucket).Object(versionFile).NewWriter(ctx)
	w.ACL = publicACL(proj)
	w.CacheControl = "no-cache"
	w.ContentType = "text/plain"
	if _, err := io.Copy(w, strings.NewReader(rev())); err != nil {
		log.Fatalf("io.Copy: %v", err)
	}
	if err := w.Close(); err != nil {
		log.Fatalf("closing GCS storage writer: %v", err)
	}
}

func TestAll(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}
	ctx := context.Background()
	ts := testutil.TokenSource(ctx, ScopePubSub, ScopeCloudPlatform)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	now := time.Now()
	topicName := fmt.Sprintf("topic-%d", now.Unix())
	subName := fmt.Sprintf("subscription-%d", now.Unix())

	client, err := NewClient(ctx, testutil.ProjID(), option.WithTokenSource(ts))
	if err != nil {
		t.Fatalf("Creating client error: %v", err)
	}
	defer client.Close()

	var topic *Topic
	if topic, err = client.CreateTopic(ctx, topicName); err != nil {
		t.Errorf("CreateTopic error: %v", err)
	}

	var sub *Subscription
	if sub, err = client.CreateSubscription(ctx, subName, topic, 0, nil); err != nil {
		t.Errorf("CreateSub error: %v", err)
	}

	exists, err := topic.Exists(ctx)
	if err != nil {
		t.Fatalf("TopicExists error: %v", err)
	}
	if !exists {
		t.Errorf("topic %s should exist, but it doesn't", topic)
	}

	exists, err = sub.Exists(ctx)
	if err != nil {
		t.Fatalf("SubExists error: %v", err)
	}
	if !exists {
		t.Errorf("subscription %s should exist, but it doesn't", subName)
	}

	msgs := []*Message{}
	for i := 0; i < 10; i++ {
		text := fmt.Sprintf("a message with an index %d", i)
		attrs := make(map[string]string)
		attrs["foo"] = "bar"
		msgs = append(msgs, &Message{
			Data:       []byte(text),
			Attributes: attrs,
		})
	}

	ids, err := topic.Publish(ctx, msgs...)
	if err != nil {
		t.Fatalf("Publish (1) error: %v", err)
	}

	if len(ids) != len(msgs) {
		t.Errorf("unexpected number of message IDs received; %d, want %d", len(ids), len(msgs))
	}

	want := make(map[string]*messageData)
	for i, m := range msgs {
		md := extractMessageData(m)
		md.ID = ids[i]
		want[md.ID] = md
	}

	// Use a timeout to ensure that Pull does not block indefinitely if there are unexpectedly few messages available.
	timeoutCtx, _ := context.WithTimeout(ctx, time.Minute)
	it, err := sub.Pull(timeoutCtx)
	if err != nil {
		t.Fatalf("error constructing iterator: %v", err)
	}
	defer it.Stop()
	got := make(map[string]*messageData)
	for i := 0; i < len(want); i++ {
		m, err := it.Next()
		if err != nil {
			t.Fatalf("error getting next message: %v", err)
		}
		md := extractMessageData(m)
		got[md.ID] = md
		m.Done(true)
	}

	if !reflect.DeepEqual(got, want) {
		t.Errorf("messages: got: %v ; want: %v", got, want)
	}

	// base64 test
	data := "[email protected]~"
	_, err = topic.Publish(ctx, &Message{Data: []byte(data)})
	if err != nil {
		t.Fatalf("Publish error: %v", err)
//.........这里部分代码省略.........

// listDownloads lists all the files found in the monthly repo, and from them,
// builds the data that we'll feed to the template to generate the monthly
// downloads camweb page.
func listDownloads() (*ReleaseData, error) {
	ts, err := tokenSource(bucket)
	if err != nil {
		return nil, err
	}
	ctx := context.Background()
	stoClient, err := storage.NewClient(ctx, option.WithTokenSource(ts), option.WithHTTPClient(oauth2.NewClient(ctx, ts)))
	if err != nil {
		return nil, err
	}
	objList, err := stoClient.Bucket(bucket).List(ctx, &storage.Query{Prefix: "monthly/"})
	if err != nil {
		return nil, err
	}

	platformBySuffix := map[string]string{
		"src.zip":       "Source",
		"linux.tar.gz":  "Linux",
		"darwin.tar.gz": "Darwin",
		"windows.zip":   "Windows",
	}
	getPlatform := func(name string) string {
		for suffix, platform := range platformBySuffix {
			if strings.HasSuffix(name, suffix) {
				return platform
			}
		}
		return ""
	}
	getChecksum := func(name string) (string, error) {
		r, err := stoClient.Bucket(bucket).Object(name).NewReader(ctx)
		if err != nil {
			return "", err
		}
		var buf bytes.Buffer
		if _, err := io.Copy(&buf, r); err != nil {
			return "", err
		}
		return buf.String(), nil
	}
	var date time.Time
	checkDate := func(objDate time.Time) error {
		if date.IsZero() {
			date = objDate
			return nil
		}
		d := date.Sub(objDate)
		if d < 0 {
			d = -d
		}
		if d < 24*time.Hour {
			return nil
		}
		return fmt.Errorf("objects in monthly have not been uploaded or updated the same day")
	}

	var (
		downloadData []DownloadData
		nameToSum    = make(map[string]string)
	)
	fileDate := releaseDate.Format(fileDateFormat)
	for _, attrs := range objList.Results {
		if !strings.Contains(attrs.Name, fileDate) {
			continue
		}
		if err := checkDate(attrs.Updated); err != nil {
			return nil, err
		}
		if !strings.HasSuffix(attrs.Name, ".sha256") {
			continue
		}
		sum, err := getChecksum(attrs.Name)
		if err != nil {
			return nil, err
		}
		nameToSum[strings.TrimSuffix(attrs.Name, ".sha256")] = sum
	}
	for _, attrs := range objList.Results {
		if !strings.Contains(attrs.Name, fileDate) {
			continue
		}
		if strings.HasSuffix(attrs.Name, ".sha256") {
			continue
		}
		sum, ok := nameToSum[attrs.Name]
		if !ok {
			return nil, fmt.Errorf("%v has no checksum file!", attrs.Name)
		}
		downloadData = append(downloadData, DownloadData{
			Filename: filepath.Base(attrs.Name),
			Platform: getPlatform(attrs.Name),
			Checksum: sum,
		})
	}

	return &ReleaseData{
		Date:         releaseDate.Format(titleDateFormat),
//.........这里部分代码省略.........

// WithTokenSource returns a ClientOption that specifies an OAuth2 token
// source to be used as the basis for authentication.
func WithTokenSource(s oauth2.TokenSource) ClientOption {
	return wrapOpt{option.WithTokenSource(s)}
}

func TestIntegration(t *testing.T) {
	if testing.Short() {
		t.Skip("Integration tests skipped in short mode")
	}

	ctx := context.Background()
	ts := testutil.TokenSource(ctx, Scope)
	if ts == nil {
		t.Skip("Integration tests skipped. See CONTRIBUTING.md for details")
	}

	projID := testutil.ProjID()
	c, err := NewClient(ctx, projID, option.WithTokenSource(ts))
	if err != nil {
		t.Fatal(err)
	}
	ds := c.Dataset("bigquery_integration_test")
	if err := ds.Create(ctx); err != nil && !hasStatusCode(err, http.StatusConflict) { // AlreadyExists is 409
		t.Fatal(err)
	}
	schema := Schema([]*FieldSchema{
		{Name: "name", Type: StringFieldType},
		{Name: "num", Type: IntegerFieldType},
	})
	table := ds.Table("t1")
	// Delete the table in case it already exists. (Ignore errors.)
	table.Delete(ctx)
	// Create the table.
	err = table.Create(ctx, schema, TableExpiration(time.Now().Add(5*time.Minute)))
	if err != nil {
		t.Fatal(err)
	}
	// Check table metadata.
	md, err := table.Metadata(ctx)
	if err != nil {
		t.Fatal(err)
	}
	// TODO(jba): check md more thorougly.
	if got, want := md.ID, fmt.Sprintf("%s:%s.%s", projID, ds.id, table.TableID); got != want {
		t.Errorf("metadata.ID: got %q, want %q", got, want)
	}
	if got, want := md.Type, RegularTable; got != want {
		t.Errorf("metadata.Type: got %v, want %v", got, want)
	}

	// List tables in the dataset.
	tables, err := ds.ListTables(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if got, want := len(tables), 1; got != want {
		t.Fatalf("ListTables: got %d, want %d", got, want)
	}
	want := *table
	if got := tables[0]; !reflect.DeepEqual(got, &want) {
		t.Errorf("ListTables: got %v, want %v", got, &want)
	}

	// Iterate over tables in the dataset.
	it := ds.Tables(ctx)
	tables = nil
	for {
		tbl, err := it.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			t.Fatal(err)
		}
		tables = append(tables, tbl)
	}
	if got, want := tables, []*Table{table}; !reflect.DeepEqual(got, want) {
		t.Errorf("Tables: got %v, want %v", got, want)
	}

	// Populate the table.
	upl := table.NewUploader()
	var rows []*ValuesSaver
	for i, name := range []string{"a", "b", "c"} {
		rows = append(rows, &ValuesSaver{
			Schema:   schema,
			InsertID: name,
			Row:      []Value{name, i},
		})
	}
	if err := upl.Put(ctx, rows); err != nil {
		t.Fatal(err)
	}

	checkRead := func(src ReadSource) {
		it, err := c.Read(ctx, src)
		if err != nil {
			t.Fatal(err)
		}
		for i := 0; it.Next(ctx); i++ {
			var vals ValueList
			if err := it.Get(&vals); err != nil {
				t.Fatal(err)
			}
			if got, want := vals, rows[i].Row; !reflect.DeepEqual([]Value(got), want) {
//.........这里部分代码省略.........