Add UT cases for package opm (update travis yaml to start redis container)

.travis.yml

@@ -28,11 +28,13 @@ env:
   AUTH_MODE: db_auth
   SELF_REGISTRATION: on
   KEY_PATH: /data/secretkey
+  REDIS_HOST: localhost
 
 before_install:
    - sudo ./tests/hostcfg.sh
    - sudo ./tests/generateCerts.sh
    - sudo ./make/prepare
+   - sudo mkdir -p "/data/redis"
 
 install: 
   - sudo apt-get update && sudo apt-get install -y libldap2-dev 
@@ -114,6 +116,7 @@ script:
   - ./tests/swaggerchecker.sh
   - ./tests/startuptest.sh
   - ./tests/userlogintest.sh ${HARBOR_ADMIN} ${HARBOR_ADMIN_PASSWD}
+  - export REDIS_HOST=$IP
 
 #  - sudo ./tests/testprepare.sh
 #  - go test -v ./tests/apitests

src/jobservice_v2/opm/redis_job_stats_mgr.go

@@ -153,6 +153,8 @@ func (rjs *RedisJobStatsManager) loop() {
 				if err := rjs.process(item); err != nil {
 					item.fails++
 					if item.fails < maxFails {
+						logger.Warningf("Failed to process '%s' request with error: %s\n", item.op, err)
+
 						//Retry after a random interval
 						go func() {
 							timer := time.NewTimer(time.Duration(backoff(item.fails)) * time.Second)

src/jobservice_v2/opm/redis_job_stats_mgr_test.go

@@ -1,6 +1,269 @@
 // Copyright 2018 The Harbor Authors. All rights reserved.
 package opm
 
-import "testing"
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"testing"
+	"time"
 
-func TestRetrieveJob(t *testing.T) {}
+	"github.com/garyburd/redigo/redis"
+	"github.com/vmware/harbor/src/jobservice_v2/job"
+	"github.com/vmware/harbor/src/jobservice_v2/models"
+	"github.com/vmware/harbor/src/jobservice_v2/utils"
+)
+
+const (
+	dialConnectionTimeout = 30 * time.Second
+	healthCheckPeriod     = time.Minute
+	dialReadTimeout       = healthCheckPeriod + 10*time.Second
+	dialWriteTimeout      = 10 * time.Second
+	testingRedisHost      = "REDIS_HOST"
+	testingNamespace      = "testing_job_service_v2"
+)
+
+var redisHost = getRedisHost()
+var redisPool = &redis.Pool{
+	MaxActive: 2,
+	MaxIdle:   2,
+	Wait:      true,
+	Dial: func() (redis.Conn, error) {
+		return redis.Dial(
+			"tcp",
+			fmt.Sprintf("%s:%d", redisHost, 6379),
+			redis.DialConnectTimeout(dialConnectionTimeout),
+			redis.DialReadTimeout(dialReadTimeout),
+			redis.DialWriteTimeout(dialWriteTimeout),
+		)
+	},
+}
+
+func TestSetJobStatus(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+	//make sure data existing
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+	<-time.After(200 * time.Millisecond)
+
+	mgr.SetJobStatus("fake_job_ID", "running")
+	<-time.After(100 * time.Millisecond)
+	stats, err := mgr.Retrieve("fake_job_ID")
+	if err != nil {
+		t.Error(err)
+	}
+
+	if stats.Stats.Status != "running" {
+		t.Errorf("expect job status 'running' but got '%s'\n", stats.Stats.Status)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestCommand(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	if err := mgr.SendCommand("fake_job_ID", CtlCommandStop); err != nil {
+		t.Error(err)
+	}
+
+	if cmd, err := mgr.CtlCommand("fake_job_ID"); err != nil {
+		t.Error(err)
+	} else {
+		if cmd != CtlCommandStop {
+			t.Errorf("expect '%s' but got '%s'", CtlCommandStop, cmd)
+		}
+	}
+
+	key := utils.KeyJobCtlCommands(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestDieAt(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+
+	dieAt := time.Now().Unix()
+	if err := createDeadJob(redisPool.Get(), dieAt); err != nil {
+		t.Error(err)
+	}
+	<-time.After(200 * time.Millisecond)
+	mgr.DieAt("fake_job_ID", dieAt)
+	<-time.After(300 * time.Millisecond)
+
+	stats, err := mgr.Retrieve("fake_job_ID")
+	if err != nil {
+		t.Error(err)
+	}
+
+	if stats.Stats.DieAt != dieAt {
+		t.Errorf("expect die at '%d' but got '%d'\n", dieAt, stats.Stats.DieAt)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+	key2 := utils.RedisKeyDead(testingNamespace)
+	if err := clear(key2, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestRegisterHook(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	if err := mgr.RegisterHook("fake_job_ID", "http://localhost:9999", false); err != nil {
+		t.Error(err)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestExpireJobStats(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	//make sure data existing
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+	<-time.After(200 * time.Millisecond)
+
+	if err := mgr.ExpirePeriodicJobStats("fake_job_ID"); err != nil {
+		t.Error(err)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func TestCheckIn(t *testing.T) {
+	mgr := createStatsManager(redisPool)
+	mgr.Start()
+	defer mgr.Shutdown()
+	<-time.After(200 * time.Millisecond)
+
+	//make sure data existing
+	testingStats := createFakeStats()
+	mgr.Save(testingStats)
+	<-time.After(200 * time.Millisecond)
+
+	//Start http server
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Fprintln(w, "ok")
+	}))
+	defer ts.Close()
+
+	if err := mgr.RegisterHook("fake_job_ID", ts.URL, false); err != nil {
+		t.Error(err)
+	}
+
+	mgr.CheckIn("fake_job_ID", "checkin")
+	<-time.After(200 * time.Millisecond)
+
+	stats, err := mgr.Retrieve("fake_job_ID")
+	if err != nil {
+		t.Error(err)
+	}
+
+	if stats.Stats.CheckIn != "checkin" {
+		t.Errorf("expect check in info 'checkin' but got '%s'\n", stats.Stats.CheckIn)
+	}
+
+	key := utils.KeyJobStats(testingNamespace, "fake_job_ID")
+	if err := clear(key, redisPool.Get()); err != nil {
+		t.Error(err)
+	}
+}
+
+func getRedisHost() string {
+	redisHost := os.Getenv(testingRedisHost)
+	if redisHost == "" {
+		redisHost = "10.160.178.186" //for local test
+	}
+
+	return redisHost
+}
+
+func createStatsManager(redisPool *redis.Pool) JobStatsManager {
+	ctx := context.Background()
+	return NewRedisJobStatsManager(ctx, testingNamespace, redisPool)
+}
+
+func clear(key string, conn redis.Conn) error {
+	if conn != nil {
+		defer conn.Close()
+		_, err := conn.Do("DEL", key)
+		return err
+	}
+
+	return errors.New("failed to clear")
+}
+
+func createFakeStats() models.JobStats {
+	testingStats := models.JobStats{
+		Stats: &models.JobStatData{
+			JobID:       "fake_job_ID",
+			JobKind:     job.JobKindPeriodic,
+			JobName:     "fake_job",
+			Status:      "Pending",
+			IsUnique:    false,
+			RefLink:     "/api/v1/jobs/fake_job_ID",
+			CronSpec:    "5 * * * * *",
+			EnqueueTime: time.Now().Unix(),
+			UpdateTime:  time.Now().Unix(),
+		},
+	}
+
+	return testingStats
+}
+
+func createDeadJob(conn redis.Conn, dieAt int64) error {
+	dead := make(map[string]interface{})
+	dead["name"] = "fake_job"
+	dead["id"] = "fake_job_ID"
+	dead["args"] = make(map[string]interface{})
+	dead["fails"] = 3
+	dead["err"] = "testing error"
+	dead["failed_at"] = dieAt
+
+	rawJSON, err := json.Marshal(&dead)
+	if err != nil {
+		return err
+	}
+
+	defer conn.Close()
+	key := utils.RedisKeyDead(testingNamespace)
+	_, err = conn.Do("ZADD", key, dieAt, rawJSON)
+	return err
+}

tests/docker-compose.test.yml

@@ -34,3 +34,10 @@ services:
       - /data/:/data/
     ports:
       - 8888:8080
+  redis:
+    image: vmware/redis-photon:4.0
+    restart: always
+    volumes:
+      - /data/redis:/data
+    ports:
+      - 6379:6379