<commit_before>package axe

import (
	"testing"
	"time"

	"github.com/globalsign/mgo/bson"
	"github.com/stretchr/testify/assert"
)

type data struct {
	Foo string `bson:"foo"`
}

func TestPool(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "foo",
		Model: &data{},
		Queue: q,
		Handler: func(m Model) (bson.M, error) {
			close(done)
			return bson.M{
				"foo": "bar",
			}, nil
		},
		Workers:     2,
		MaxAttempts: 2,
	})
	p.Run()

	time.Sleep(100 * time.Millisecond)

	job, err := q.Enqueue("foo", &data{Foo: "bar"}, 0)
	assert.NoError(t, err)

	<-done

	time.Sleep(100 * time.Millisecond)

	job = tester.Fetch(&Job{}, job.ID()).(*Job)
	assert.Equal(t, "foo", job.Name)
	assert.Equal(t, &data{Foo: "bar"}, decodeRaw(job.Data, &data{}))
	assert.Equal(t, StatusCompleted, job.Status)
	assert.NotZero(t, job.Created)
	assert.NotZero(t, job.Available)
	assert.NotZero(t, job.Started)
	assert.NotZero(t, job.Ended)
	assert.NotZero(t, job.Finished)
	assert.Equal(t, 1, job.Attempts)
	assert.Equal(t, bson.M{"foo": "bar"}, job.Result)
	assert.Equal(t, "", job.Reason)

	p.Close()
}

func TestPoolDelayed(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "delayed",
		Model: &data{},
		Queue: q,
		Handler: func(m Model) (bson.M, error) {
			close(done)
			return bson.M{
				"foo": "bar",
			}, nil
		},
		Workers:     2,
		MaxAttempts: 2,
	})
	p.Run()

	time.Sleep(100 * time.Millisecond)

	job, err := q.Enqueue("delayed", &data{Foo: "bar"}, 100*time.Millisecond)
	assert.NoError(t, err)

	<-done

	time.Sleep(100 * time.Millisecond)

	job = tester.Fetch(&Job{}, job.ID()).(*Job)
	assert.Equal(t, "delayed", job.Name)
	assert.Equal(t, &data{Foo: "bar"}, decodeRaw(job.Data, &data{}))
	assert.Equal(t, StatusCompleted, job.Status)
	assert.NotZero(t, job.Created)
	assert.NotZero(t, job.Available)
	assert.NotZero(t, job.Started)
	assert.NotZero(t, job.Ended)
	assert.NotZero(t, job.Finished)
	assert.Equal(t, 1, job.Attempts)
	assert.Equal(t, bson.M{"foo": "bar"}, job.Result)
	assert.Equal(t, "", job.Reason)

	p.Close()
}

func TestPoolFailed(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	i := 0

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "failed",
		Model: &data{},
		Queue: q,
		Handler: func(m Model) (bson.M, error) {
			if i == 0 {
				i++
				return nil, E("foo", true)
			} else {
				close(done)
				return bson.M{
					"foo": "bar",
				}, nil
			}
		},
		Workers:     2,
		MaxAttempts: 2,
	})
	p.Run()

	time.Sleep(100 * time.Millisecond)

	job, err := q.Enqueue("failed", &data{Foo: "bar"}, 0)
	assert.NoError(t, err)

	<-done

	time.Sleep(100 * time.Millisecond)

	job = tester.Fetch(&Job{}, job.ID()).(*Job)
	assert.Equal(t, "failed", job.Name)
	assert.Equal(t, &data{Foo: "bar"}, decodeRaw(job.Data, &data{}))
	assert.Equal(t, StatusCompleted, job.Status)
	assert.NotZero(t, job.Created)
	assert.NotZero(t, job.Available)
	assert.NotZero(t, job.Started)
	assert.NotZero(t, job.Ended)
	assert.NotZero(t, job.Finished)
	assert.Equal(t, 2, job.Attempts)
	assert.Equal(t, bson.M{"foo": "bar"}, job.Result)
	assert.Equal(t, "foo", job.Reason)

	p.Close()
}

func TestPoolCancel(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "cancel",<commit_msg>added test for crashing jobs<commit_after>package axe

import (
	"io"
	"testing"
	"time"

	"github.com/globalsign/mgo/bson"
	"github.com/stretchr/testify/assert"
)

type data struct {
	Foo string `bson:"foo"`
}

func TestPool(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "foo",
		Model: &data{},
		Queue: q,
		Handler: func(m Model) (bson.M, error) {
			close(done)
			return bson.M{
				"foo": "bar",
			}, nil
		},
		Workers:     2,
		MaxAttempts: 2,
	})
	p.Run()

	time.Sleep(100 * time.Millisecond)

	job, err := q.Enqueue("foo", &data{Foo: "bar"}, 0)
	assert.NoError(t, err)

	<-done

	time.Sleep(100 * time.Millisecond)

	job = tester.Fetch(&Job{}, job.ID()).(*Job)
	assert.Equal(t, "foo", job.Name)
	assert.Equal(t, &data{Foo: "bar"}, decodeRaw(job.Data, &data{}))
	assert.Equal(t, StatusCompleted, job.Status)
	assert.NotZero(t, job.Created)
	assert.NotZero(t, job.Available)
	assert.NotZero(t, job.Started)
	assert.NotZero(t, job.Ended)
	assert.NotZero(t, job.Finished)
	assert.Equal(t, 1, job.Attempts)
	assert.Equal(t, bson.M{"foo": "bar"}, job.Result)
	assert.Equal(t, "", job.Reason)

	p.Close()
}

func TestPoolDelayed(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "delayed",
		Model: &data{},
		Queue: q,
		Handler: func(m Model) (bson.M, error) {
			close(done)
			return bson.M{
				"foo": "bar",
			}, nil
		},
		Workers:     2,
		MaxAttempts: 2,
	})
	p.Run()

	time.Sleep(100 * time.Millisecond)

	job, err := q.Enqueue("delayed", &data{Foo: "bar"}, 100*time.Millisecond)
	assert.NoError(t, err)

	<-done

	time.Sleep(100 * time.Millisecond)

	job = tester.Fetch(&Job{}, job.ID()).(*Job)
	assert.Equal(t, "delayed", job.Name)
	assert.Equal(t, &data{Foo: "bar"}, decodeRaw(job.Data, &data{}))
	assert.Equal(t, StatusCompleted, job.Status)
	assert.NotZero(t, job.Created)
	assert.NotZero(t, job.Available)
	assert.NotZero(t, job.Started)
	assert.NotZero(t, job.Ended)
	assert.NotZero(t, job.Finished)
	assert.Equal(t, 1, job.Attempts)
	assert.Equal(t, bson.M{"foo": "bar"}, job.Result)
	assert.Equal(t, "", job.Reason)

	p.Close()
}

func TestPoolFailed(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	i := 0

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "failed",
		Model: &data{},
		Queue: q,
		Handler: func(m Model) (bson.M, error) {
			if i == 0 {
				i++
				return nil, E("foo", true)
			} else {
				close(done)
				return bson.M{
					"foo": "bar",
				}, nil
			}
		},
		Workers:     2,
		MaxAttempts: 2,
	})
	p.Run()

	time.Sleep(100 * time.Millisecond)

	job, err := q.Enqueue("failed", &data{Foo: "bar"}, 0)
	assert.NoError(t, err)

	<-done

	time.Sleep(100 * time.Millisecond)

	job = tester.Fetch(&Job{}, job.ID()).(*Job)
	assert.Equal(t, "failed", job.Name)
	assert.Equal(t, &data{Foo: "bar"}, decodeRaw(job.Data, &data{}))
	assert.Equal(t, StatusCompleted, job.Status)
	assert.NotZero(t, job.Created)
	assert.NotZero(t, job.Available)
	assert.NotZero(t, job.Started)
	assert.NotZero(t, job.Ended)
	assert.NotZero(t, job.Finished)
	assert.Equal(t, 2, job.Attempts)
	assert.Equal(t, bson.M{"foo": "bar"}, job.Result)
	assert.Equal(t, "foo", job.Reason)

	p.Close()
}

func TestPoolCrashed(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	i := 0

	p := NewPool()
	// p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "crashed",
		Model: &data{},
		Queue: q,
		Handler: func(m Model) (bson.M, error) {
			if i == 0 {
				i++
				return nil, io.EOF
			} else {
				close(done)
				return nil, nil
			}
		},
		Workers:     2,
		MaxAttempts: 2,
	})
	p.Run()

	time.Sleep(100 * time.Millisecond)

	job, err := q.Enqueue("crashed", &data{}, 0)
	assert.NoError(t, err)

	<-done

	time.Sleep(100 * time.Millisecond)

	job = tester.Fetch(&Job{}, job.ID()).(*Job)
	assert.Equal(t, "crashed", job.Name)
	assert.Equal(t, &data{}, decodeRaw(job.Data, &data{}))
	assert.Equal(t, StatusCompleted, job.Status)
	assert.NotZero(t, job.Created)
	assert.NotZero(t, job.Available)
	assert.NotZero(t, job.Started)
	assert.NotZero(t, job.Ended)
	assert.NotZero(t, job.Finished)
	assert.Equal(t, 2, job.Attempts)
	assert.Equal(t, bson.M{}, job.Result)
	assert.Equal(t, "EOF", job.Reason)

	p.Close()
}

func TestPoolCancel(t *testing.T) {
	tester.Clean()

	q := NewQueue(tester.Store)

	done := make(chan struct{})

	p := NewPool()
	p.Reporter = func(err error) { panic(err) }
	p.Add(&Task{
		Name:  "cancel",
{'fraction_non_alphanumeric': 0.19822945328388564, 'fraction_numerical': 0.009570522789807393, 'mean_word_length': 3.777142857142857, 'pattern_counts': {'":': 12, '<': 10, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4391881', 'n_tokens_mistral': 3800, 'n_tokens_neox': 3457, 'n_words': 691}