/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package pool

import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/rs/zerolog"
	"github.com/stretchr/testify/assert"
)

func Test_newExecutor(t *testing.T) {
	type args struct {
		queueDepth             int
		numberOfInitialWorkers int
		log                    zerolog.Logger
	}
	tests := []struct {
		name        string
		args        args
		want        *executor
		manipulator func(t *testing.T, want *executor, got *executor)
	}{
		{
			name: "just create it",
			want: &executor{
				worker: []*worker{},
			},
			manipulator: func(t *testing.T, want *executor, got *executor) {
				assert.NotNil(t, got.workItems)
				want.workItems = got.workItems
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := newExecutor(tt.args.queueDepth, tt.args.numberOfInitialWorkers, tt.args.log)
			want := tt.want
			if tt.manipulator != nil {
				tt.manipulator(t, want, got)
			}
			assert.Equalf(t, want, got, "newExecutor(%v, %v, %v)", tt.args.queueDepth, tt.args.numberOfInitialWorkers, tt.args.log)
		})
	}
}

func Test_executor_Close(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		workItems    chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name    string
		fields  fields
		wantErr assert.ErrorAssertionFunc
	}{
		{
			name:    "close it",
			wantErr: assert.NoError,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.workItems,
				traceWorkers: tt.fields.traceWorkers,
				log:          produceTestingLogger(t),
			}
			tt.wantErr(t, e.Close(), fmt.Sprintf("Close()"))
		})
	}
}

func Test_executor_IsRunning(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		workItems    chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name   string
		fields fields
		want   bool
	}{
		{
			name: "no",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.workItems,
				traceWorkers: tt.fields.traceWorkers,
				log:          produceTestingLogger(t),
			}
			assert.Equalf(t, tt.want, e.IsRunning(), "IsRunning()")
		})
	}
}

func Test_executor_Start(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		queue        chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name      string
		fields    fields
		shouldRun bool
	}{
		{
			name:      "Start fresh",
			shouldRun: true,
		},
		{
			name: "Start running",
			fields: fields{
				running: true,
			},
			shouldRun: true,
		},
		{
			name: "Start stopping",
			fields: fields{
				running:  true,
				shutdown: true,
			},
			shouldRun: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.queue,
				traceWorkers: tt.fields.traceWorkers,
			}
			e.Start()
			assert.Equal(t, tt.shouldRun, e.IsRunning(), "should be running")
		})
	}
}

func Test_executor_Stop(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		queue        chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name      string
		fields    fields
		shouldRun bool
	}{
		{
			name:      "Stop stopped",
			shouldRun: false,
		},
		{
			name: "Stop running",
			fields: fields{
				running: true,
				queue:   make(chan workItem),
				worker: []*worker{
					func() *worker {
						w := &worker{}
						w.initialize()
						return w
					}(),
				},
			},
			shouldRun: false,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.queue,
				traceWorkers: tt.fields.traceWorkers,
			}
			e.Stop()
		})
	}
}

func Test_executor_Submit(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		queue        chan workItem
		traceWorkers bool
	}
	type args struct {
		workItemId int32
		runnable   Runnable
		context    context.Context
	}
	tests := []struct {
		name                      string
		fields                    fields
		args                      args
		completionFutureValidator func(t *testing.T, future CompletionFuture) bool
		waitForCompletion         bool
	}{
		{
			name: "submitting nothing",
			completionFutureValidator: func(t *testing.T, completionFuture CompletionFuture) bool {
				return assert.Error(t, completionFuture.(*future).err.Load().(error))
			},
		},
		{
			name: "submit canceled",
			fields: fields{
				queue: make(chan workItem),
			},
			args: args{
				workItemId: 13,
				runnable: func() {
					// We do something for 3 seconds
					<-time.NewTimer(3 * time.Second).C
				},
				context: func() context.Context {
					ctx, cancelFunc := context.WithCancel(context.Background())
					cancelFunc()
					return ctx
				}(),
			},
			completionFutureValidator: func(t *testing.T, completionFuture CompletionFuture) bool {
				err := completionFuture.(*future).err.Load().(error)
				return assert.Error(t, err)
			},
		},
		{
			name: "Submit something which doesn't complete",
			fields: fields{
				queue: make(chan workItem, 1),
			},
			args: args{
				workItemId: 13,
				runnable: func() {
					// We do something for 3 seconds
					<-time.NewTimer(3 * time.Second).C
				},
				context: t.Context(),
			},
			completionFutureValidator: func(t *testing.T, completionFuture CompletionFuture) bool {
				completed := completionFuture.(*future).completed.Load()
				return assert.False(t, completed)
			},
		},
		{
			name: "Submit something which does complete",
			fields: func() fields {
				var executor = NewFixedSizeExecutor(1, 1).(*executor)
				return fields{
					running:      executor.running,
					shutdown:     executor.shutdown,
					worker:       executor.worker,
					queue:        executor.workItems,
					traceWorkers: true,
				}
			}(),
			args: args{
				workItemId: 13,
				runnable: func() {
					// NOOP
				},
				context: t.Context(),
			},
			completionFutureValidator: func(t *testing.T, completionFuture CompletionFuture) bool {
				completed := completionFuture.(*future).completed.Load()
				return assert.True(t, completed)
			},
			waitForCompletion: true,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.queue,
				traceWorkers: tt.fields.traceWorkers,
			}
			e.Start()
			completionFuture := e.Submit(tt.args.context, tt.args.workItemId, tt.args.runnable)
			if tt.waitForCompletion {
				assert.NoError(t, completionFuture.AwaitCompletion(testContext(t)))
			}
			assert.True(t, tt.completionFutureValidator(t, completionFuture), "Submit(%v, %v)", tt.args.workItemId, tt.args.runnable)
		})
	}
}

func Test_executor_getWorkerWaitGroup(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		workItems    chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name   string
		fields fields
		want   *sync.WaitGroup
	}{
		{
			name: "get it",
			want: &sync.WaitGroup{},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.workItems,
				traceWorkers: tt.fields.traceWorkers,
				log:          produceTestingLogger(t),
			}
			assert.Equalf(t, tt.want, e.getWorkerWaitGroup(), "getWorkerWaitGroup()")
		})
	}
}

func Test_executor_getWorksItems(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		workItems    chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name   string
		fields fields
		want   chan workItem
	}{
		{
			name: "get it",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.workItems,
				traceWorkers: tt.fields.traceWorkers,
				log:          produceTestingLogger(t),
			}
			assert.Equalf(t, tt.want, e.getWorksItems(), "getWorksItems()")
		})
	}
}

func Test_executor_isTraceWorkers(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		workItems    chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name   string
		fields fields
		want   bool
	}{
		{
			name: "it is not",
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.workItems,
				traceWorkers: tt.fields.traceWorkers,
				log:          produceTestingLogger(t),
			}
			assert.Equalf(t, tt.want, e.isTraceWorkers(), "isTraceWorkers()")
		})
	}
}

func Test_executor_String(t *testing.T) {
	type fields struct {
		running      bool
		shutdown     bool
		worker       []*worker
		workItems    chan workItem
		traceWorkers bool
	}
	tests := []struct {
		name   string
		fields fields
		want   string
	}{
		{
			name: "string it",
			fields: fields{
				running:  true,
				shutdown: true,
				worker: []*worker{
					{
						id:          1,
						shutdown:    atomic.Bool{},
						interrupted: atomic.Bool{},
						lastReceived: func() atomic.Value {
							value := atomic.Value{}
							value.Store(time.Time{})
							return value
						}(),
					},
				},
				traceWorkers: true,
			},
			want: `
╔═executor═════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
║╔═running╗╔═shutdown╗                                                                                                 ║
║║b1 true ║║ b1 true ║                                                                                                 ║
║╚════════╝╚═════════╝                                                                                                 ║
║╔═worker/value/worker══════════════════════════════════════════════════════════════════════════════════╗╔═workItems══╗║
║║╔═id═════════════════╗╔═lastReceived════════════════╗╔═running╗╔═shutdown╗╔═interrupted╗╔═interrupter╗║║0 element(s)║║
║║║0x0000000000000001 1║║0001-01-01 00:00:00 +0000 UTC║║b0 false║║b0 false ║║  b0 false  ║║0 element(s)║║╚════════════╝║
║║╚════════════════════╝╚═════════════════════════════╝╚════════╝╚═════════╝╚════════════╝╚════════════╝║              ║
║╚══════════════════════════════════════════════════════════════════════════════════════════════════════╝              ║
║╔═traceWorkers╗                                                                                                       ║
║║   b1 true   ║                                                                                                       ║
║╚═════════════╝                                                                                                       ║
╚══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝`[1:],
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			e := &executor{
				running:      tt.fields.running,
				shutdown:     tt.fields.shutdown,
				worker:       tt.fields.worker,
				workItems:    tt.fields.workItems,
				traceWorkers: tt.fields.traceWorkers,
				log:          produceTestingLogger(t),
			}
			assert.Equalf(t, tt.want, e.String(), "String()")
		})
	}
}
