func TestTCPParser(t *testing.T) {
	parserVersion := parser.InitParserVersionForTest()
	taskfilename := "testdata/20190516T013026.744845Z-tcpinfo-mlab4-arn02-ndt.tgz"
	url := "gs://fake-archive/ndt/tcpinfo/2019/05/16/" + filepath.Base(taskfilename)
	src, err := fileSource(taskfilename)
	if err != nil {
		t.Fatal("Failed reading testdata from", taskfilename)
	}
	ins := newInMemorySink()
	p := parser.NewTCPInfoParser(ins, "test", "_suffix", newFakeAnnotator(tcpInfoAnno))
	task := task.NewTask(url, src, p, nullCloser{})
	startDecode := time.Now()
	n, err := task.ProcessAllTests(false)
	decodeTime := time.Since(startDecode)
	if err != nil {
		t.Fatal(err)
	}
	if n != 364 {
		t.Errorf("Expected ProcessAllTests to handle %d files, but it handled %d.\n", 364, n)
	}
	if ins.Committed() != 362 {
		t.Errorf("Expected %d, Got %d.", 362, ins.Committed())
	}
	if len(ins.data) < 1 {
		t.Fatal("Should have at least one inserted row")
	}
	for i, rawRow := range ins.data {
		row, ok := rawRow.(*schema.TCPRow)
		if !ok {
			t.Fatal("not a TCPRow")
		}
		if row.ParseInfo.ParseTime.After(time.Now()) {
			t.Error("Should have inserted parse_time")
		}
		if row.ParseInfo.TaskFileName != url {
			t.Error("Should have correct taskfilename", taskfilename, "!=", row.ParseInfo.TaskFileName)
		}
		if !strings.Contains(row.ParseInfo.Filename, row.UUID) {
			t.Errorf("Should have non empty filename containing UUID: %s not found in :%s:\n", row.UUID, row.ParseInfo.Filename)
		}
		if row.ParseInfo.ParserVersion != parserVersion {
			t.Error("ParserVersion not properly set", row.ParseInfo.ParserVersion)
		}
		if i < 5 && row.SockID.SPort != 3010 {
			t.Error("SPort should be 3010", row.SockID, i)
		}
		if row.SockID.SrcIP != "195.89.146.242" && row.SockID.SrcIP != "2001:5012:100:24::242" {
			t.Error("Wrong SrcIP", row.SockID.SrcIP)
		}
		if row.Client == nil {
			t.Error("Client annotations should not be nil", row.SockID, row.FinalSnapshot)
		}
		if row.Server == nil {
			t.Error("Server annotations should not be nil")
		} else if row.Server.IATA == "" {
			t.Error("Server IATA should not be empty")
		}
	}
	startMarshal := time.Now()
	var largestRow *schema.TCPRow
	var largestJson []byte
	totalSnaps := int64(0)
	for _, r := range ins.data {
		row, _ := r.(*schema.TCPRow)
		jsonBytes, _ := json.Marshal(r)
		totalSnaps += int64(len(row.Snapshots))
		if len(jsonBytes) > len(largestJson) {
			largestRow = row
			largestJson = jsonBytes
		}
	}
	marshalTime := time.Since(startMarshal)
	duration := largestRow.FinalSnapshot.Timestamp.Sub(largestRow.Snapshots[0].Timestamp)
	t.Log("Largest json is", len(largestJson), "bytes in", len(largestRow.Snapshots), "snapshots, over", duration, "with", len(largestJson)/len(largestRow.Snapshots), "json bytes/snap")
	t.Log("Total of", totalSnaps, "snapshots decoded and marshalled")
	t.Log("Average", decodeTime.Nanoseconds()/totalSnaps, "nsec/snap to decode", marshalTime.Nanoseconds()/totalSnaps, "nsec/snap to marshal")
	snapJson, _ := json.Marshal(largestRow.FinalSnapshot)
	t.Log(string(snapJson))
	if duration > 20*time.Second {
		t.Error("Incorrect duration calculation", duration)
	}
	if totalSnaps != 1588 {
		t.Error("expected 1588 (thinned) snapshots, got", totalSnaps)
	}
	cx := v2.ConvertAnnotationsToClientAnnotations(tcpInfoAnno["35.225.75.192"])
	if diff := deep.Equal(&largestRow.ClientX, cx); diff != nil {
		t.Errorf("ClientX annotation does not match; %#v", diff)
	}
	sx := v2.ConvertAnnotationsToServerAnnotations(tcpInfoAnno["195.89.146.242"])
	sx.Site = "arn02"
	sx.Machine = "mlab4"
	if diff := deep.Equal(&largestRow.ServerX, sx); diff != nil {
		t.Errorf("ServerX annotation does not match; %#v", diff)
	}
}
NOTE: This uses a fake annotator which returns no annotations.
TODO: This test seems to be flakey in travis - sometimes only 357 tests instead of 362
package parser_test
import (
	"archive/tar"
	"compress/gzip"
	"context"
	"encoding/json"
	"errors"
	"io"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"
	"github.com/go-test/deep"
	v2 "github.com/m-lab/annotation-service/api/v2"
	"cloud.google.com/go/bigquery"
	"cloud.google.com/go/civil"
	"github.com/m-lab/annotation-service/api"
	"github.com/m-lab/etl/etl"
	"github.com/m-lab/etl/parser"
	"github.com/m-lab/etl/schema"
	"github.com/m-lab/etl/storage"
	"github.com/m-lab/etl/task"
)
func assertTCPInfoParser(in *parser.TCPInfoParser) {
	func(p etl.Parser) {}(in)
}
func fileSource(fn string) (etl.TestSource, error) {
	if !(strings.HasSuffix(fn, ".tgz") || strings.HasSuffix(fn, ".tar") ||
		strings.HasSuffix(fn, ".tar.gz")) {
		return nil, errors.New("not tar or tgz: " + fn)
	}
	var rdr io.ReadCloser
	var raw io.ReadCloser
	raw, err := os.Open(fn)
	if err != nil {
		return nil, err
	}
	if strings.HasSuffix(strings.ToLower(fn), "gz") {
		rdr, err = gzip.NewReader(raw)
		if err != nil {
			raw.Close()
			return nil, err
		}
	} else {
		rdr = raw
	}
	tarReader := tar.NewReader(rdr)
	timeout := 16 * time.Millisecond
	return &storage.GCSSource{TarReader: tarReader, Closer: raw,
		RetryBaseTime: timeout, TableBase: "test", PathDate: civil.Date{Year: 2020, Month: 6, Day: 11}}, nil
}
var tcpInfoAnno = map[string]*api.Annotations{
	"35.225.75.192": &api.Annotations{
		Geo: &api.GeolocationIP{
			ContinentCode: "NA",
			CountryCode:   "US",
			Latitude:      1.0,
			Longitude:     2.0,
		},
		Network: &api.ASData{
			ASNumber: 1234,
			Systems: []api.System{
				{ASNs: []uint32{1234}},
			},
		},
	},
	"195.89.146.242": &api.Annotations{
		Geo: &api.GeolocationIP{
			ContinentCode: "NA",
			CountryCode:   "US",
			Latitude:      1.0,
			Longitude:     2.0,
		},
		Network: &api.ASData{
			ASNumber: 1234,
			Systems: []api.System{
				{ASNs: []uint32{1234}},
			},
		},
	},
}
type inMemorySink struct {
	data      []interface{}
	committed int
	failed    int
	token     chan struct{}
}
func newInMemorySink() *inMemorySink {
	data := make([]interface{}, 0)
	token := make(chan struct{}, 1)
	token <- struct{}{}
	return &inMemorySink{data, 0, 0, token}
}
func (in *inMemorySink) acquire() {
	<-in.token
}
func (in *inMemorySink) release() {
	in.token <- struct{}{}  
}
func (in *inMemorySink) Commit(data []interface{}, label string) (int, error) {
	in.acquire()
	defer in.release()
	in.data = append(in.data, data...)
	in.committed = len(in.data)
	return len(data), nil
}
func (in *inMemorySink) Close() error {
	return nil
}
func (in *inMemorySink) Flush() error {
	in.committed = len(in.data)
	return nil
}
func (in *inMemorySink) Committed() int {
	return in.committed
}
type nullCloser struct{}
func (nc nullCloser) Close() error { return nil }