func TestGetGSResultFileLocations(t *testing.T) {
	testutils.SkipIfShort(t)
	storage, err := storage.New(http.DefaultClient)
	assert.Nil(t, err)

	startTS := time.Date(2014, time.December, 10, 0, 0, 0, 0, time.UTC).Unix()
	endTS := time.Date(2014, time.December, 10, 23, 59, 59, 0, time.UTC).Unix()

	// TODO(stephana): Switch this to a dedicated test bucket, so we are not
	// in danger of removing it.
	resultFiles, err := getGSResultsFileLocations(startTS, endTS, storage, "chromium-skia-gm", "dm-json-v1")
	assert.Nil(t, err)

	// Read the expected list of files and compare them.
	content, err := ioutil.ReadFile("./testdata/filelist_dec_10.txt")
	assert.Nil(t, err)
	lines := strings.Split(strings.TrimSpace(string(content)), "\n")
	sort.Strings(lines)

	resultNames := make([]string, len(resultFiles))
	for idx, rf := range resultFiles {
		resultNames[idx] = rf.Name
	}
	sort.Strings(resultNames)
	assert.Equal(t, len(lines), len(resultNames))
	assert.Equal(t, lines, resultNames)
}

func TestGoogleStorageSource(t *testing.T) {
	testutils.SkipIfShort(t)

	src, err := NewGoogleStorageSource("gs-test-src", TEST_GS_BUCKET, TEST_GS_DIR, http.DefaultClient)
	assert.Nil(t, err)
	testSource(t, src)
}

// testBuildKeyOrdering ensures that we properly sort build keys so that the
// build numbers are strictly ascending.
func testBuildKeyOrdering(t *testing.T, local bool) {
	testutils.SkipIfShort(t)
	d := clearDB(t, local)
	defer d.Close(t)

	b := "Test-Builder"
	m := "test.master"
	assert.Nil(t, d.DB().PutBuild(&Build{
		Builder: b,
		Master:  m,
		Number:  1,
	}))
	assert.Nil(t, d.DB().PutBuild(&Build{
		Builder: b,
		Master:  m,
		Number:  10,
	}))
	assert.Nil(t, d.DB().PutBuild(&Build{
		Builder: b,
		Master:  m,
		Number:  2,
	}))
	max, err := d.DB().GetMaxBuildNumber(m, b)
	assert.Nil(t, err)
	assert.Equal(t, 10, max)
}

func TestGetLocalResultFileLocations(t *testing.T) {
	testutils.SkipIfShort(t)

	err := testutils.DownloadTestDataArchive(t, TEST_DATA_STORAGE_PATH, TEST_DATA_DIR)
	assert.Nil(t, err)

	startTS := time.Date(2015, time.May, 5, 0, 0, 0, 0, time.UTC).Unix()
	endTS := time.Date(2015, time.May, 17, 23, 59, 59, 0, time.UTC).Unix()

	resultFiles, err := getLocalResultsFileLocations(startTS, endTS, filepath.Join(TEST_DATA_DIR, "nano-json-v1"))
	assert.Nil(t, err)

	// Read the expected list of files and compare them.
	content, err := ioutil.ReadFile("./testdata/local_ingest_files.txt")
	assert.Nil(t, err)
	lines := strings.Split(strings.TrimSpace(string(content)), "\n")
	sort.Strings(lines)

	resultNames := make([]string, len(resultFiles))
	for idx, rf := range resultFiles {
		resultNames[idx] = rf.Name
	}
	sort.Strings(resultNames)
	assert.Equal(t, len(lines), len(resultNames))
	assert.Equal(t, lines, resultNames)
}

// testBuildDbSerialization verifies that we can write a build to the DB and
// pull it back out without losing or corrupting the data.
func testBuildDbSerialization(t *testing.T, local bool) {
	testutils.SkipIfShort(t)

	d := clearDB(t, local)
	defer d.Close(t)

	// Load the test repo.
	tr := util.NewTempRepo()
	defer tr.Cleanup()

	repos := gitinfo.NewRepoMap(tr.Dir)

	// Test case: an empty build. Tests null and empty values.
	emptyBuild := &Build{
		Steps:   []*BuildStep{},
		Commits: []string{},
	}
	emptyBuild.fixup()

	// Test case: a completely filled-out build.
	buildFromFullJson, err := testGetBuildFromMaster(repos)
	assert.Nil(t, err)

	testCases := []*Build{emptyBuild, buildFromFullJson}
	for _, b := range testCases {
		dbSerializeAndCompare(t, d, b, true)
	}
}

// TestGetModeHistory verifies that we correctly track mode history.
func TestGetModeHistory(t *testing.T) {
	testutils.SkipIfShort(t)
	d := newTestDB(t)
	defer d.cleanup(t)

	// Single mode.
	m1 := &ModeChange{
		Message: "Starting!",
		Mode:    MODE_RUNNING,
		Time:    time.Now().UTC(),
		User:    "[email protected]",
	}
	assert.Nil(t, d.db.SetMode(m1))
	history, err := d.db.GetModeHistory(10)
	assert.Nil(t, err)
	assert.Equal(t, 1, len(history))
	testutils.AssertDeepEqual(t, m1, history[0])

	// Add more modes, ensuring that we retrieve them consistently.
	m2 := &ModeChange{
		Message: "Stoppit",
		Mode:    MODE_STOPPED,
		Time:    time.Now().UTC().Add(time.Minute),
		User:    "[email protected]",
	}
	m3 := &ModeChange{
		Message: "Dry run",
		Mode:    MODE_DRY_RUN,
		Time:    time.Now().UTC().Add(2 * time.Minute),
		User:    "[email protected]",
	}
	m4 := &ModeChange{
		Message: "Dry run",
		Mode:    MODE_DRY_RUN,
		Time:    time.Now().UTC().Add(3 * time.Minute),
		User:    "[email protected]",
	}

	assert.Nil(t, d.db.SetMode(m2))
	history, err = d.db.GetModeHistory(10)
	assert.Nil(t, err)
	testutils.AssertDeepEqual(t, []*ModeChange{m2, m1}, history)

	assert.Nil(t, d.db.SetMode(m3))
	history, err = d.db.GetModeHistory(10)
	assert.Nil(t, err)
	testutils.AssertDeepEqual(t, []*ModeChange{m3, m2, m1}, history)

	assert.Nil(t, d.db.SetMode(m4))
	history, err = d.db.GetModeHistory(10)
	assert.Nil(t, err)
	testutils.AssertDeepEqual(t, []*ModeChange{m4, m3, m2, m1}, history)

	// Only three changes?
	history, err = d.db.GetModeHistory(3)
	assert.Nil(t, err)
	testutils.AssertDeepEqual(t, []*ModeChange{m4, m3, m2}, history)
}

func testBuildQueue(t *testing.T, timeDecay24Hr float64, expectations []*buildQueueExpect, testInsert bool) {
	testutils.SkipIfShort(t)

	// Initialize the buildbot database.
	d := clearDB(t)
	defer d.Close(t)

	// Load the test repo.
	tr := util.NewTempRepo()
	defer tr.Cleanup()
	repos := gitinfo.NewRepoMap(tr.Dir)
	repo, err := repos.Repo(TEST_REPO)
	assert.Nil(t, err)
	assert.Nil(t, repos.Update())

	// Create the BuildQueue.
	q, err := NewBuildQueue(PERIOD_FOREVER, repos, DEFAULT_SCORE_THRESHOLD, timeDecay24Hr, []*regexp.Regexp{}, d.db)
	assert.Nil(t, err)

	// Fake time.Now()
	details, err := repo.Details(hashes['I'], false)
	assert.Nil(t, err)
	now := details.Timestamp.Add(1 * time.Hour)

	// Update the queue.
	assert.Nil(t, q.update(now))

	// Ensure that we get the expected BuildCandidate at each step. Insert
	// each BuildCandidate into the buildbot database to simulate actually
	// running builds.
	buildNum := 0
	for _, expected := range expectations {
		bc, err := q.Pop([]string{TEST_BUILDER})
		assert.Equal(t, expected.err, err)
		if err != nil {
			break
		}
		glog.Infof("\n%v\n%v", expected.bc, bc)
		assert.True(t, reflect.DeepEqual(expected.bc, bc))
		if testInsert || buildNum == 0 {
			// Actually insert a build, as if we're really using the scheduler.
			// Do this even if we're not testing insertion, because if we don't,
			// the queue won't know about this builder.
			b := &buildbot.Build{
				Builder:     bc.Builder,
				Master:      "fake",
				Number:      buildNum,
				BuildSlave:  "fake",
				Branch:      "master",
				GotRevision: bc.Commit,
				Repository:  TEST_REPO,
			}
			assert.Nil(t, buildbot.IngestBuild(d.db, b, repos))
			buildNum++
			assert.Nil(t, q.update(now))
		}
	}
}

// testUnfinishedBuild verifies that we can write a build which is not yet
// finished, load the build back from the database, and update it when it
// finishes.
func testUnfinishedBuild(t *testing.T, local bool) {
	testutils.SkipIfShort(t)
	d := clearDB(t, local)
	defer d.Close(t)

	// Load the test repo.
	tr := util.NewTempRepo()
	defer tr.Cleanup()

	repos := gitinfo.NewRepoMap(tr.Dir)

	// Obtain and insert an unfinished build.
	httpClient = testHttpClient
	b, err := getBuildFromMaster("client.skia", "Test-Ubuntu12-ShuttleA-GTX550Ti-x86_64-Release-Valgrind", 152, repos)
	assert.Nil(t, err)
	assert.False(t, b.IsFinished(), "Unfinished build thinks it's finished!")
	dbSerializeAndCompare(t, d, b, true)

	// Ensure that the build is found by GetUnfinishedBuilds.
	unfinished, err := d.DB().GetUnfinishedBuilds(b.Master)
	assert.Nil(t, err)
	found := false
	for _, u := range unfinished {
		if u.Master == b.Master && u.Builder == b.Builder && u.Number == b.Number {
			found = true
			break
		}
	}
	assert.True(t, found, "Unfinished build was not found by getUnfinishedBuilds!")

	// Add another step to the build to "finish" it, ensure that we can
	// retrieve it as expected.
	b.Finished = b.Started.Add(30 * time.Second)
	stepStarted := b.Started.Add(500 * time.Millisecond)
	s := &BuildStep{
		Name:     "LastStep",
		Number:   len(b.Steps),
		Results:  0,
		Started:  stepStarted,
		Finished: b.Finished,
	}
	b.Steps = append(b.Steps, s)
	assert.True(t, b.IsFinished(), "Finished build thinks it's unfinished!")
	dbSerializeAndCompare(t, d, b, true)

	// Ensure that the finished build is NOT found by getUnfinishedBuilds.
	unfinished, err = d.DB().GetUnfinishedBuilds(b.Master)
	assert.Nil(t, err)
	found = false
	for _, u := range unfinished {
		if u.Master == b.Master && u.Builder == b.Builder && u.Number == b.Number {
			found = true
			break
		}
	}
	assert.False(t, found, "Finished build was found by getUnfinishedBuilds!")
}

// TestModeHistory verifies that we correctly track mode history.
func TestModeHistory(t *testing.T) {
	testutils.SkipIfShort(t)

	// Create the ModeHistory.
	tmpDir, err := ioutil.TempDir("", "test_autoroll_mode_")
	assert.Nil(t, err)
	defer testutils.RemoveAll(t, tmpDir)
	mh, err := NewModeHistory(path.Join(tmpDir, "test.db"))
	assert.Nil(t, err)
	defer func() {
		assert.Nil(t, mh.Close())
	}()

	// Use this function for checking expectations.
	check := func(expect, actual []*ModeChange) {
		assert.Equal(t, len(expect), len(actual))
		for i, e := range expect {
			assert.Equal(t, e.Mode, actual[i].Mode)
			assert.Equal(t, e.Message, actual[i].Message)
			assert.Equal(t, e.User, actual[i].User)
		}

	}

	// Initial mode, set automatically.
	mc0 := &ModeChange{
		Message: "Setting initial mode.",
		Mode:    MODE_RUNNING,
		User:    "AutoRoll Bot",
	}

	expect := []*ModeChange{mc0}
	setModeAndCheck := func(mc *ModeChange) {
		assert.Nil(t, mh.Add(mc.Mode, mc.User, mc.Message))
		assert.Equal(t, mc.Mode, mh.CurrentMode())
		expect = append([]*ModeChange{mc}, expect...)
		check(expect, mh.GetHistory())
	}

	// Ensure that we set our initial state properly.
	assert.Equal(t, mc0.Mode, mh.CurrentMode())
	check(expect, mh.GetHistory())

	// Change the mode.
	setModeAndCheck(&ModeChange{
		Message: "Stop the presses!",
		Mode:    MODE_STOPPED,
		User:    "[email protected]",
	})

	// Change a few times.
	setModeAndCheck(&ModeChange{
		Message: "Resume!",
		Mode:    MODE_RUNNING,
		User:    "[email protected]",
	})
}

func TestBlamerWithLiveData(t *testing.T) {
	testutils.SkipIfShort(t)

	err := testutils.DownloadTestDataFile(t, TEST_DATA_STORAGE_PATH, TEST_DATA_PATH)
	assert.Nil(t, err, "Unable to download testdata.")
	defer testutils.RemoveAll(t, TEST_DATA_DIR)

	tileStore := mocks.NewMockTileStoreFromJson(t, TEST_DATA_PATH)
	testBlamerWithLiveData(t, tileStore)
}

func TestStatusWatcher(t *testing.T) {
	testutils.SkipIfShort(t)

	err := testutils.DownloadTestDataFile(t, TEST_DATA_STORAGE_PATH, TEST_DATA_PATH)
	assert.Nil(t, err, "Unable to download testdata.")
	defer testutils.RemoveAll(t, TEST_DATA_DIR)

	tileBuilder := mocks.NewMockTileBuilderFromJson(t, TEST_DATA_PATH)
	testStatusWatcher(t, tileBuilder)
}

func TestFileSystemResultFileLocations(t *testing.T) {
	testutils.SkipIfShort(t)

	err := testutils.DownloadTestDataArchive(t, TEST_DATA_STORAGE_PATH, TEST_DATA_DIR)
	assert.Nil(t, err)
	defer testutils.RemoveAll(t, TEST_DATA_DIR)

	src, err := NewFileSystemSource("test-fs-source", TEST_DATA_DIR)
	assert.Nil(t, err)
	testSource(t, src)
}

func testBuildQueue(t *testing.T, timeDecay24Hr float64, expectations []*buildQueueExpect) {
	testutils.SkipIfShort(t)

	// Initialize the buildbot database.
	d := clearDB(t)
	defer d.Close(t)

	// Load the test repo.
	tr := util.NewTempRepo()
	defer tr.Cleanup()
	repos := gitinfo.NewRepoMap(tr.Dir)
	repo, err := repos.Repo("/skia.git")
	assert.Nil(t, err)
	assert.Nil(t, repos.Update())

	// Create the BuildQueue.
	q, err := NewBuildQueue(PERIOD_FOREVER, tr.Dir, DEFAULT_SCORE_THRESHOLD, timeDecay24Hr, []string{TEST_BUILDER})
	assert.Nil(t, err)

	// Fake time.Now()
	details, err := repo.Details(hashes['I'])
	assert.Nil(t, err)
	now := details.Timestamp.Add(1 * time.Hour)

	// Ensure that we get the expected BuildCandidate at each step. Insert
	// each BuildCandidate into the buildbot database to simulate actually
	// running builds.
	buildNum := 0
	for _, expected := range expectations {
		assert.Nil(t, q.update(now))

		bc, err := q.Pop(TEST_BUILDER)
		assert.Equal(t, expected.err, err)
		if err != nil {
			break
		}
		hash, err := repo.FullHash(bc.Commit)
		assert.Nil(t, err)
		bc.Commit = hash
		assert.True(t, reflect.DeepEqual(expected.bc, bc))
		b := &buildbot.Build{
			Builder:     bc.Builder,
			Master:      "fake",
			Number:      buildNum,
			BuildSlave:  "fake",
			Branch:      "master",
			GotRevision: bc.Commit,
			Repository:  TEST_REPO,
		}
		assert.Nil(t, buildbot.IngestBuild(b, repos))
		buildNum++
	}
}

// TODO(stephana): Disable until fixed.
func TestEventBus(t *testing.T) {
	testutils.SkipIfShort(t)

	eventBus, err := NewNSQEventBus("127.0.0.1:4150")
	assert.Nil(t, err)

	ch := make(chan string, 100)
	var wg sync.WaitGroup

	callbackFn := func(ready *int32) func([]byte) {
		return func(data []byte) {
			if string(data) == "ready" {
				atomic.StoreInt32(ready, 0)
				return
			}
			ch <- string(data)
			wg.Done()
		}
	}

	var ready_1 int32 = -1
	var ready_2 int32 = -1
	var ready_3 int32 = -1
	assert.Nil(t, eventBus.SubscribeAsync("topic1", callbackFn(&ready_1)))
	assert.Nil(t, eventBus.SubscribeAsync("topic2", callbackFn(&ready_2)))
	assert.Nil(t, eventBus.SubscribeAsync("topic2", callbackFn(&ready_3)))

	for atomic.LoadInt32(&ready_1)+atomic.LoadInt32(&ready_2)+atomic.LoadInt32(&ready_3) < 0 {
		assert.Nil(t, eventBus.Publish("topic1", []byte("ready")))
		assert.Nil(t, eventBus.Publish("topic2", []byte("ready")))
		time.Sleep(time.Millisecond)
	}

	wg.Add(3)
	assert.Nil(t, eventBus.Publish("topic1", []byte("0")))
	assert.Nil(t, eventBus.Publish("topic2", []byte("msg-01")))
	wg.Wait()

	assert.True(t, len(ch) >= 3)
	close(ch)

	vals := []string{}
	for val := range ch {
		vals = append(vals, val)
	}

	sort.Strings(vals)
	assert.Equal(t, []string{"0", "msg-01", "msg-01"}, vals)

	assert.Nil(t, eventBus.Close())
}

// TestGetBuildFromMaster verifies that we can load JSON data from the build master and
// decode it into a Build object.
func TestGetBuildFromMaster(t *testing.T) {
	testutils.SkipIfShort(t)

	// Load the test repo.
	tr := util.NewTempRepo()
	defer tr.Cleanup()

	repos := gitinfo.NewRepoMap(tr.Dir)

	// Default, complete build.
	_, err := testGetBuildFromMaster(repos)
	assert.Nil(t, err)
	// Incomplete build.
	_, err = getBuildFromMaster("client.skia", "Test-Ubuntu12-ShuttleA-GTX550Ti-x86_64-Release-Valgrind", 152, repos)
	assert.Nil(t, err)
}

// Basic test to make sure we can retrieve issues from Rietveld.
func TestRietveld(t *testing.T) {
	testutils.SkipIfShort(t)

	api := New("https://codereview.chromium.org", nil)
	t_delta := time.Now().Add(-10 * 24 * time.Hour)
	issues, err := api.Search(1, SearchModifiedAfter(t_delta))
	assert.Nil(t, err)
	assert.True(t, len(issues) > 0)

	for _, issue := range issues {
		assert.True(t, issue.Modified.After(t_delta))
		details, err := api.GetIssueProperties(issue.Issue, false)
		assert.Nil(t, err)
		assert.True(t, details.Modified.After(t_delta))
		assert.True(t, len(details.Patchsets) > 0)
	}
}

// TestBuildJsonSerialization verifies that we can serialize a build to JSON
// and back without losing or corrupting the data.
func TestBuildJsonSerialization(t *testing.T) {
	testutils.SkipIfShort(t)

	// Load the test repo.
	tr := util.NewTempRepo()
	defer tr.Cleanup()

	repos := gitinfo.NewRepoMap(tr.Dir)

	b1, err := testGetBuildFromMaster(repos)
	assert.Nil(t, err)
	bytes, err := json.Marshal(b1)
	assert.Nil(t, err)
	b2 := &Build{}
	assert.Nil(t, json.Unmarshal(bytes, b2))
	testutils.AssertDeepEqual(t, b1, b2)
}

func TestEventBusGlobally(t *testing.T) {
	testutils.SkipIfShort(t)

	globalEventBus, err := geventbus.NewNSQEventBus(NSQD_ADDR)
	assert.Nil(t, err)

	secondGlobalBus, err := geventbus.NewNSQEventBus(NSQD_ADDR)
	assert.Nil(t, err)

	eventBus := New(globalEventBus)
	ch := make(chan interface{}, 100)
	eventBus.SubscribeAsync(GLOBAL_TOPIC, func(e interface{}) { ch <- e })

	secondCh := make(chan interface{}, 100)
	errCh := make(chan error, 100)
	assert.Nil(t, secondGlobalBus.SubscribeAsync(GLOBAL_TOPIC, geventbus.JSONCallback(&testType{}, func(data interface{}, err error) {
		if err != nil {
			errCh <- err
		} else {
			secondCh <- data
		}
	})))

	eventBus.Publish(GLOBAL_TOPIC, &testType{0, "message-1"})
	eventBus.Publish(GLOBAL_TOPIC, &testType{1, "message-2"})
	eventBus.Publish(GLOBAL_TOPIC, &testType{2, "message-3"})
	eventBus.Publish(GLOBAL_TOPIC, &testType{3, "message-4"})
	time.Sleep(5 * time.Second)
	assert.Equal(t, 4, len(ch))
	assert.Equal(t, 4, len(secondCh))
	close(ch)
	close(secondCh)

	found := map[int]bool{}
	for m := range ch {
		assert.IsType(t, &testType{}, m)
		temp := m.(*testType)
		assert.False(t, found[temp.ID])
		found[temp.ID] = true
	}

	for m := range secondCh {
		assert.IsType(t, &testType{}, m)
	}
}

func TestCompareSources(t *testing.T) {
	testutils.SkipIfShort(t)

	gsSource, err := NewGoogleStorageSource("gs-test-src", TEST_GS_BUCKET, TEST_GS_DIR, http.DefaultClient)
	assert.Nil(t, err)

	err = testutils.DownloadTestDataArchive(t, TEST_DATA_STORAGE_PATH, TEST_DATA_DIR)
	assert.Nil(t, err)
	defer testutils.RemoveAll(t, TEST_DATA_DIR)

	fsSource, err := NewFileSystemSource("test-fs-source", TEST_DATA_DIR)
	assert.Nil(t, err)

	gsResults, err := gsSource.Poll(START_TIME, END_TIME)
	assert.Nil(t, err)

	fsResults, err := fsSource.Poll(START_TIME, END_TIME)
	assert.Nil(t, err)

	assert.Equal(t, len(gsResults), len(fsResults))
	sort.Sort(rflSlice(gsResults))
	sort.Sort(rflSlice(fsResults))

	// Only compare a subset of all files to limit the runtime of this test.
	gsResults = gsResults[:10]
	fsResults = fsResults[:10]

	for idx, result := range gsResults {
		// Check if the MD5 from GS and the file system match.
		assert.Equal(t, result.MD5(), fsResults[idx].MD5())

		// Check if the content is the same for both sources.
		file, err := result.Open()
		assert.Nil(t, err)
		c1, err := ioutil.ReadAll(file)
		assert.Nil(t, err)
		file, err = fsResults[idx].Open()
		c2, err := ioutil.ReadAll(file)
		assert.Nil(t, err)
		assert.Equal(t, string(c1), string(c2))
	}
}

// TestAlertDBSerialization verifies that we properly serialize and
// deserialize Alerts into the DB.
func TestAlertDBSerialization(t *testing.T) {
	testutils.SkipIfShort(t)
	d := clearDB(t)
	defer d.Close(t)

	cases := []*Alert{
		&Alert{},    // Empty Alert.
		makeAlert(), // Realistic case.
	}

	for _, want := range cases {
		assert.Nil(t, want.retryReplaceIntoDB())
		a, err := GetActiveAlerts()
		assert.Nil(t, err)
		assert.Equal(t, 1, len(a))
		got := a[0]
		testutils.AssertDeepEqual(t, got, want)
		// Dismiss the Alert, so that it doesn't show up later.
		got.DismissedAt = 1000
		assert.Nil(t, got.retryReplaceIntoDB())
	}
}