Merge pull request #264 from RTradeLtd/queue/workers-2

Queue Cleanup + Additional Tests: TEM-126

.travis.yml

@@ -4,7 +4,9 @@ go:
 services:
   - docker
 sudo: required
-
+env:
+  - SSL_MODE_DISABLE=true
+
 install:
   - docker pull koalaman/shellcheck
   - go get -u golang.org/x/lint/golint

Makefile

@@ -50,6 +50,8 @@ testenv:
 	( cd testenv ; make testenv )
 	@echo "Running migrations..."
 	@env CONFIG_DAG=./testenv/config.json go run cmd/temporal/main.go migrate-insecure
+	make api-user
+	make api-admin
 	@echo "===================          done           ==================="
 
 # Shut down testenv

api/api.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/RTradeLtd/ChainRider-Go/dash"
 	clients "github.com/RTradeLtd/Temporal/grpc-clients"
+	"github.com/RTradeLtd/Temporal/queue"
 	"github.com/RTradeLtd/rtfs"
 
 	limit "github.com/aviddiviner/gin-limit"
@@ -37,6 +38,10 @@ var (
 	dev      = true
 )
 
+const (
+	realName = "temporal-realm"
+)
+
 // API is our API service
 type API struct {
 	ipfs    rtfs.Manager
@@ -57,6 +62,7 @@ type API struct {
 	orch    *clients.IPFSOrchestratorClient
 	lc      *clients.LensClient
 	dc      *dash.Client
+	queues  queues
 	service string
 }
 
@@ -140,7 +146,7 @@ func new(cfg *config.TemporalConfig, router *gin.Engine, ipfs rtfs.Manager, debu
 		return nil, err
 	}
 
-	orch, err := clients.NewOcrhestratorClient(cfg.Orchestrator, os.Getenv("MODE") == "development")
+	orch, err := clients.NewOcrhestratorClient(cfg.Orchestrator)
 	if err != nil {
 		return nil, err
 	}
@@ -168,6 +174,43 @@ func new(cfg *config.TemporalConfig, router *gin.Engine, ipfs rtfs.Manager, debu
 	if err != nil {
 		return nil, err
 	}
+	// setup our queues
+	qmIpns, err := queue.New(queue.IpnsEntryQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmPin, err := queue.New(queue.IpfsPinQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmDatabase, err := queue.New(queue.DatabaseFileAddQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmFile, err := queue.New(queue.IpfsFileQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmCluster, err := queue.New(queue.IpfsClusterPinQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmEmail, err := queue.New(queue.EmailSendQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmKey, err := queue.New(queue.IpfsKeyCreationQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmDash, err := queue.New(queue.DashPaymentConfirmationQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
+	qmPayConfirm, err := queue.New(queue.PaymentConfirmationQueue, cfg.RabbitMQ.URL, true)
+	if err != nil {
+		return nil, err
+	}
 	return &API{
 		ipfs:    ipfs,
 		keys:    keys,
@@ -185,17 +228,51 @@ func new(cfg *config.TemporalConfig, router *gin.Engine, ipfs rtfs.Manager, debu
 		signer:  signer,
 		orch:    orch,
 		dc:      dc,
-		zm:      models.NewZoneManager(dbm.DB),
-		rm:      models.NewRecordManager(dbm.DB),
-		nm:      models.NewHostedIPFSNetworkManager(dbm.DB),
+		queues: queues{
+			pin:        qmPin,
+			file:       qmFile,
+			cluster:    qmCluster,
+			email:      qmEmail,
+			ipns:       qmIpns,
+			key:        qmKey,
+			database:   qmDatabase,
+			dash:       qmDash,
+			payConfirm: qmPayConfirm,
+		},
+		zm: models.NewZoneManager(dbm.DB),
+		rm: models.NewRecordManager(dbm.DB),
+		nm: models.NewHostedIPFSNetworkManager(dbm.DB),
 	}, nil
 }
 
 // Close releases API resources
 func (api *API) Close() {
+	// close grpc connections
 	api.lc.Close()
 	api.signer.Close()
 	api.orch.Close()
+	// close queue resources
+	if err := api.queues.cluster.Close(); err != nil {
+		api.LogError(err, "failed to properly close cluster queue connection")
+	}
+	if err := api.queues.database.Close(); err != nil {
+		api.LogError(err, "failed to properly close database queue connection")
+	}
+	if err := api.queues.email.Close(); err != nil {
+		api.LogError(err, "failed to properly close email queue connection")
+	}
+	if err := api.queues.file.Close(); err != nil {
+		api.LogError(err, "failed to properly close file queue connection")
+	}
+	if err := api.queues.ipns.Close(); err != nil {
+		api.LogError(err, "failed to properly close ipns queue connection")
+	}
+	if err := api.queues.key.Close(); err != nil {
+		api.LogError(err, "failed to properly close key queue connection")
+	}
+	if err := api.queues.pin.Close(); err != nil {
+		api.LogError(err, "failed to properly close pin queue connection")
+	}
 }
 
 // TLSConfig is used to enable TLS on the API service
@@ -245,7 +322,7 @@ func (api *API) setupRoutes() error {
 		stats.RequestStats())
 
 	// set up middleware
-	ginjwt := middleware.JwtConfigGenerate(api.cfg.API.JwtKey, api.dbm.DB, api.l)
+	ginjwt := middleware.JwtConfigGenerate(api.cfg.API.JwtKey, realName, api.dbm.DB, api.l)
 	authware := []gin.HandlerFunc{
 		ginjwt.MiddlewareFunc(),
 		middleware.APIRestrictionMiddleware(api.dbm.DB),

api/api_test.go

@@ -1,27 +1,101 @@
 package api
 
 import (
-	"os"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
 	"testing"
-	"time"
 
-	"github.com/RTradeLtd/config"
-	"github.com/RTradeLtd/rtfs"
 	"github.com/gin-gonic/gin"
+
+	"github.com/RTradeLtd/config"
+	"github.com/c2h5oh/datasize"
+)
+
+const (
+	tooManyCredits = 10.9999997e+07
+	testUser       = "testuser"
 )
 
-func Test_new(t *testing.T) {
+func Test_API(t *testing.T) {
 	cfg, err := config.LoadConfig("../testenv/config.json")
 	if err != nil {
 		t.Fatal(err)
 	}
-	ipfs, err := rtfs.NewManager(cfg.IPFS.APIConnection.Host+":"+cfg.IPFS.APIConnection.Port, nil, time.Minute*5)
+	api, err := Initialize(cfg, true)
 	if err != nil {
 		t.Fatal(err)
 	}
-	api, err := new(cfg, gin.New(), ipfs, true, os.Stdout)
+	if err = api.FileSizeCheck(int64(datasize.GB.Bytes() * 1)); err != nil {
+		t.Fatal(err)
+	}
+	if err = api.FileSizeCheck(int64(datasize.GB.Bytes() * 10)); err == nil {
+		t.Fatal("error expected")
+	}
+	type args struct {
+		paymentType string
+		blockchain  string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		wantErr bool
+	}{
+		{"ETHFail", args{"ETH", "ETH"}, true},
+		{"ETHPass", args{"eth", "ethereum"}, false},
+		{"BTCFail", args{"BTC", "BTC"}, true},
+		{"BTCPass", args{"btc", "bitcoin"}, false},
+		{"LTCFail", args{"LTC", "LTC"}, true},
+		{"LTCPass", args{"ltc", "litecoin"}, false},
+		{"XMRFail", args{"XMR", "XMR"}, true},
+		{"XMRPass", args{"xmr", "monero"}, false},
+		{"DASHFail", args{"DASH", "DASH"}, true},
+		{"DASHPass", args{"dash", "dash"}, false},
+		{"InvalidCoinFail", args{"biiiitcoooonnneeeeeeecccct", "biiiitcoooonnneeeeeeecccct"}, true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if _, err := api.getDepositAddress(tt.args.paymentType); (err != nil) != tt.wantErr {
+				t.Errorf("getDepositAddress() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if valid := api.validateBlockchain(tt.args.blockchain); !valid != tt.wantErr {
+				t.Errorf("validateBlockchain() error = %v, wantErr %v", valid, tt.wantErr)
+			}
+		})
+	}
+	if err = api.validateUserCredits(testUser, 1); err != nil {
+		t.Fatal(err)
+	}
+	if err = api.validateUserCredits(testUser, tooManyCredits); err == nil {
+		t.Fatal("error expected")
+	}
+	if err := api.validateAdminRequest(testUser); err != nil {
+		t.Fatal(err)
+	}
+	if err := api.validateAdminRequest("notareallaccount"); err == nil {
+		t.Fatal("error expected")
+	}
+	user, err := api.um.FindByUserName(testUser)
 	if err != nil {
 		t.Fatal(err)
 	}
-	api.setupRoutes()
+	previousCreditAmount := user.Credits
+	api.refundUserCredits(testUser, "ipfs-pin", 10)
+	user, err = api.um.FindByUserName(testUser)
+	if user.Credits != previousCreditAmount+10 {
+		t.Fatal("failed to refund credits")
+	}
+	recorder := httptest.NewRecorder()
+	testCtx, _ := gin.CreateTestContext(recorder)
+	urlValues := url.Values{}
+	urlValues.Add("suchkey", "muchvalue")
+	testCtx.Request = &http.Request{PostForm: urlValues}
+	forms := api.extractPostForms(testCtx, "suchkey")
+	if len(forms) == 0 {
+		t.Fatal("failed to extract post forms")
+	}
+	if forms["suchkey"] != "muchvalue" {
+		t.Fatal("failed to extract proper postform")
+	}
 }

api/middleware/jwt.go

@@ -10,16 +10,14 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-var realmName = "temporal-realm"
-
 // Login is used to unmarshal a login in request so that we can parse it
 type Login struct {
 	Username string `form:"username" json:"username" binding:"required"`
 	Password string `form:"password" json:"password" binding:"required"`
 }
 
 // JwtConfigGenerate is used to generate our JWT configuration
-func JwtConfigGenerate(jwtKey string, db *gorm.DB, logger *log.Logger) *jwt.GinJWTMiddleware {
+func JwtConfigGenerate(jwtKey, realmName string, db *gorm.DB, logger *log.Logger) *jwt.GinJWTMiddleware {
 
 	// will implement metamaks/msg signing with ethereum accounts
 	// as the authentication metho