Test: tasks pkg

pkg/task/compress_test.go

@@ -144,7 +144,7 @@ func TestCompressTask_Do(t *testing.T) {
 		task.Do()
 		asserts.NoError(mock.ExpectationsWereMet())
 		asserts.NotEmpty(task.GetError().Msg)
-		asserts.True(util.IsEmpty("test/compress"))
+		asserts.True(util.IsEmpty(util.RelativePath("test/compress")))
 	}
 }
 

pkg/task/job.go

@@ -82,7 +82,7 @@ func Record(job Job) (*model.Task, error) {
 }
 
 // Resume 从数据库中恢复未完成任务
-func Resume() {
+func Resume(p Pool) {
 	tasks := model.GetTasksByStatus(Queued, Processing)
 	if len(tasks) == 0 {
 		return
@@ -97,7 +97,7 @@ func Resume() {
 		}
 
 		if job != nil {
-			TaskPoll.Submit(job)
+			p.Submit(job)
 		}
 	}
 }

pkg/task/job_test.go

@@ -2,6 +2,7 @@ package task
 
 import (
 	"errors"
+	testMock "github.com/stretchr/testify/mock"
 	"testing"
 
 	"github.com/DATA-DOG/go-sqlmock"
@@ -21,14 +22,45 @@ func TestRecord(t *testing.T) {
 	asserts.NoError(err)
 }
 
+type taskPoolMock struct {
+	testMock.Mock
+}
+
+func (t taskPoolMock) Add(num int) {
+	t.Called(num)
+}
+
+func (t taskPoolMock) Submit(job Job) {
+	t.Called(job)
+}
+
 func TestResume(t *testing.T) {
 	asserts := assert.New(t)
+	mockPool := taskPoolMock{}
 
 	// 没有任务
 	{
-		mock.ExpectQuery("SELECT(.+)").WithArgs(Queued).WillReturnRows(sqlmock.NewRows([]string{"type"}))
-		Resume()
+		mock.ExpectQuery("SELECT(.+)").WithArgs(Queued, Processing).WillReturnRows(sqlmock.NewRows([]string{"type"}))
+		Resume(mockPool)
+		asserts.NoError(mock.ExpectationsWereMet())
+	}
+
+	// 有任务, 类型未知
+	{
+		mock.ExpectQuery("SELECT(.+)").WithArgs(Queued, Processing).WillReturnRows(sqlmock.NewRows([]string{"type"}).AddRow(233))
+		Resume(mockPool)
+		asserts.NoError(mock.ExpectationsWereMet())
+	}
+
+	// 有任务
+	{
+		mockPool.On("Submit", testMock.Anything)
+		mock.ExpectQuery("SELECT(.+)").WithArgs(Queued, Processing).WillReturnRows(sqlmock.NewRows([]string{"type", "props"}).AddRow(CompressTaskType, "{}"))
+		mock.ExpectQuery("SELECT(.+)users").WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
+		mock.ExpectQuery("SELECT(.+)policies").WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
+		Resume(mockPool)
 		asserts.NoError(mock.ExpectationsWereMet())
+		mockPool.AssertExpectations(t)
 	}
 }
 

pkg/task/pool.go

@@ -63,6 +63,6 @@ func Init() {
 	util.Log().Info("初始化任务队列，WorkerNum = %d", maxWorker)
 
 	if conf.SystemConfig.Mode == "master" {
-		Resume()
+		Resume(TaskPoll)
 	}
 }

pkg/task/pool_test.go

@@ -29,10 +29,10 @@ func TestMain(m *testing.M) {
 func TestInit(t *testing.T) {
 	asserts := assert.New(t)
 	cache.Set("setting_max_worker_num", "10", 0)
-	mock.ExpectQuery("SELECT(.+)").WithArgs(Queued).WillReturnRows(sqlmock.NewRows([]string{"type"}).AddRow(-1))
+	mock.ExpectQuery("SELECT(.+)").WithArgs(Queued, Processing).WillReturnRows(sqlmock.NewRows([]string{"type"}).AddRow(-1))
 	Init()
 	asserts.NoError(mock.ExpectationsWereMet())
-	asserts.Len(TaskPoll.idleWorker, 10)
+	asserts.Len(TaskPoll.(*AsyncPool).idleWorker, 10)
 }
 
 func TestPool_Submit(t *testing.T) {

pkg/task/slavetask/transfer.go

@@ -69,7 +69,7 @@ func (job *TransferTask) SetErrorMsg(msg string, err error) {
 	}
 
 	if err := cluster.DefaultController.SendNotification(job.MasterID, job.Req.Hash(job.MasterID), notifyMsg); err != nil {
-		util.Log().Warning("无法发送转存失败通知到从机, ", err)
+		util.Log().Warning("无法发送转存失败通知到从机, %s", err)
 	}
 }
 
@@ -132,7 +132,7 @@ func (job *TransferTask) Do() {
 	}
 
 	if err := cluster.DefaultController.SendNotification(job.MasterID, job.Req.Hash(job.MasterID), msg); err != nil {
-		util.Log().Warning("无法发送转存成功通知到从机, ", err)
+		util.Log().Warning("无法发送转存成功通知到从机, %s", err)
 	}
 }
 

pkg/task/transfer_test.go

@@ -138,7 +138,7 @@ func TestNewTransferTask(t *testing.T) {
 		mock.ExpectBegin()
 		mock.ExpectExec("INSERT(.+)").WillReturnResult(sqlmock.NewResult(1, 1))
 		mock.ExpectCommit()
-		job, err := NewTransferTask(1, []string{}, "/", "/", false)
+		job, err := NewTransferTask(1, []string{}, "/", "/", false, 0, nil)
 		asserts.NoError(mock.ExpectationsWereMet())
 		asserts.NotNil(job)
 		asserts.NoError(err)
@@ -150,7 +150,7 @@ func TestNewTransferTask(t *testing.T) {
 		mock.ExpectBegin()
 		mock.ExpectExec("INSERT(.+)").WillReturnError(errors.New("error"))
 		mock.ExpectRollback()
-		job, err := NewTransferTask(1, []string{}, "/", "/", false)
+		job, err := NewTransferTask(1, []string{}, "/", "/", false, 0, nil)
 		asserts.NoError(mock.ExpectationsWereMet())
 		asserts.Nil(job)
 		asserts.Error(err)