Merge pull request #3 from qa-dev/concurrency-tests

concurrency tests

Author: andrskom

.travis.yml

@@ -4,7 +4,8 @@ go:
   - 1.8.x
 
 install: make get-deps prepare
-script: go test ./...
+script:
+  - make test
 services:
   - mysql
 

Dockerfile

@@ -7,6 +7,6 @@ ENV CONFIG_PATH ./config.json
 
 RUN make
 
-CMD ["service-entrypoint"]
+CMD ["jsonwire-grid"]
 
 EXPOSE 4444

Makefile

@@ -1,4 +1,5 @@
 export CONFIG_PATH=./config.json
+TEST_CONFIG_PATH=./config-test.json
 all: get-deps build
 
 .PHONY: help build fmt clean run test coverage check vet lint doc cfpush
@@ -11,7 +12,8 @@ help:
 	@echo "run      - start application"
 
 build: prepare
-	go build -o ${GOPATH}/bin/service-entrypoint
+	@echo "Install jsonwire-grid"
+	go install github.com/qa-dev/jsonwire-grid
 
 fmt:
 	go fmt
@@ -25,4 +27,27 @@ get-deps:
 prepare: fmt gen
 
 run: build
-	${GOPATH}/bin/service-entrypoint
+	@echo "Start jsonwire-grid"
+	jsonwire-grid
+
+test:
+	go test ./...
+
+concurrency-test-prepare: build
+	@echo "Install jsonwire-grid"
+	go install github.com/qa-dev/jsonwire-grid/testing/webdriver-node-mock
+	@echo "Install webdriver-node-mock"
+	go install github.com/qa-dev/jsonwire-grid/testing/webdriver-mock-creator
+	@echo "Install webdriver-mock-creator"
+	go install github.com/qa-dev/jsonwire-grid/testing/webdriver-concurrency-test
+	@echo "Kill all running jsonwire-grid"
+	killall -9 jsonwire-grid &
+	@echo "Wait 1s"
+	sleep 1
+	@echo "Start jsonwire-grid"
+	CONFIG_PATH=$(TEST_CONFIG_PATH) jsonwire-grid &
+
+concurrency-test: concurrency-test-prepare
+	@echo "Start webdriver-concurrency-test"
+	webdriver-concurrency-test
+

config-sample.json

@@ -4,7 +4,7 @@
   },
   "db": {
       "implementation": "mysql",
-      "connection": "root:@(localhost:3306)/grid?tx_isolation=SERIALIZABLE&parseTime=true"
+      "connection": "root:@(localhost:3306)/grid?tx_isolation=SERIALIZABLE&parseTime=true&interpolateParams=true"
   },
   "grid": {
     "port": 4444,

config-test.json

@@ -0,0 +1,27 @@
+{
+  "logger": {
+    "level": 3
+  },
+  "db": {
+      "implementation": "mysql",
+      "connection": "root:@(127.0.0.1:3306)/grid?tx_isolation=SERIALIZABLE&parseTime=true&interpolateParams=true"
+  },
+  "grid": {
+    "port": 4444,
+    "strategy_list": [
+      {
+        "type": "default",
+        "limit": 0
+      }
+    ],
+    "busy_node_duration": "15m",
+    "reserved_node_duration": "5m"
+  },
+  "statsd": {
+    "host": "statsd-host",
+    "port": 8126,
+    "protocol": "udp",
+    "prefix": "products.tests.qa.debug-dev.jsonwire-grid.",
+    "enable": false
+  }
+}

main.go

@@ -79,15 +79,20 @@ func main() {
 	http.Handle("/", m.Log(&handlers.UseSession{Pool: poolInstance}))
 
 	server := &http.Server{Addr: fmt.Sprintf(":%v", cfg.Grid.Port)}
+	serverError := make(chan error)
 	go func() {
 		err = server.ListenAndServe()
 		if err != nil {
-			// todo: норма ли что при закрытии всегда возвращается еррор???
-			log.Errorf("Listen serve error, %s", err)
+			// todo: норма ли что при вызове server.Shutdown всегда возвращается еррор???
+			serverError <- err
 		}
 	}()
 
-	<-stop
+	select {
+	case err = <-serverError:
+		log.Fatalf("Server error, %s", err)
+	case <-stop:
+	}
 
 	log.Info("Shutting down the server...")
 

storage/mysql/factory.go

@@ -21,7 +21,7 @@ func (f *Factory) Create(config config.Config) (pool.StorageInterface, error) {
 		return nil, err
 	}
 
-	db.SetMaxIdleConns(0) // this is the root problem! set it to 0 to remove all idle connections
+	db.SetMaxIdleConns(0)  // this is the root problem! set it to 0 to remove all idle connections
 	db.SetMaxOpenConns(50) // or whatever is appropriate for your setup.
 
 	storage := NewMysqlStorage(db)

storage/mysql/storage.go

@@ -2,13 +2,13 @@ package mysql
 
 import (
 	"errors"
+	"fmt"
 	"github.com/jmoiron/sqlx"
 	"github.com/qa-dev/jsonwire-grid/pool"
 	"sort"
 	"strconv"
-	"time"
-	"fmt"
 	"strings"
+	"time"
 )
 
 type MysqlNodeModel struct {

testing/webdriver-concurrency-test/webdriver-concurrency-test.go

@@ -0,0 +1,192 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"github.com/qa-dev/jsonwire-grid/jsonwire"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"os"
+	"os/exec"
+	"os/signal"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+var (
+	hubUrl      *string
+	level       *int
+	durationStr *string
+)
+
+// todo: prototype
+func main() {
+	hubUrl = flag.String("hub", "http://127.0.0.1:4444", "address of hub, default http://127.0.0.1:4444")
+	level = flag.Int("level", 50, "count parallell conections")
+	durationStr = flag.String("duration", "30s", "duration of test, string format ex. 12m, see time.ParseDuration()")
+	mockMaxDuration := flag.Int("mockMaxDuration", 500, "request duration [0 <=duration], default 0")
+	mockStartPort := flag.Int("mockStartPort", 5000, "mockStartPort")
+	flag.Parse()
+
+	duration, err := time.ParseDuration(*durationStr)
+	if err != nil {
+		log.Fatal("Invalid duration")
+	}
+	var counter uint64 = 0
+
+	stop := make(chan os.Signal)
+	signal.Notify(stop, os.Interrupt)
+	wg := sync.WaitGroup{}
+	isAlive := true
+	errChan := make(chan error, *level)
+
+	cmd := exec.Command(
+		"webdriver-mock-creator",
+		fmt.Sprintf("-hub=%v", *hubUrl),
+		fmt.Sprintf("-startPort=%v", *mockStartPort),
+		fmt.Sprintf("-maxDuration=%v", *mockMaxDuration),
+		fmt.Sprintf("-countNodes=%v", *level),
+	)
+
+	err = cmd.Start()
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	time.Sleep(time.Millisecond * 500)
+
+	go func() {
+		for i := 1; i <= *level && isAlive; i++ {
+			time.Sleep(time.Millisecond * 100)
+			go func() {
+				wg.Add(1)
+				defer wg.Done()
+				for {
+					if !isAlive {
+						break
+					}
+					err := runScenario()
+					if err != nil {
+						errChan <- errors.New("Run scenario, " + err.Error())
+					}
+					atomic.AddUint64(&counter, 1)
+				}
+			}()
+		}
+	}()
+
+	select {
+	case <-time.After(duration):
+	case err = <-errChan:
+	case <-stop:
+	}
+
+	isAlive = false
+
+	//wait end all running scenarios
+	wg.Wait()
+
+	//wait interrupt child process
+	cmd.Process.Signal(os.Interrupt)
+	cmd.Wait()
+
+	if err != nil {
+		log.Fatalf("Tests failed: %v, ", err)
+	}
+	log.Printf("Test ok, %v cycles passed", counter)
+}
+
+func runScenario() error {
+	sessionID, err := createSession()
+	if err != nil {
+		err = errors.New("Create session, " + err.Error())
+		return err
+	}
+	err = sendAnyRequest(sessionID)
+	if err != nil {
+		err = errors.New("Send any request, " + err.Error())
+		return err
+	}
+	err = closeSession(sessionID)
+	if err != nil {
+		err = errors.New("Close session, " + err.Error())
+		return err
+	}
+	return nil
+}
+
+func createSession() (sessionID string, err error) {
+	resp, err := http.Post(*hubUrl+"/wd/hub/session", "application/json", bytes.NewBuffer([]byte(`{"desiredCapabilities":{"browserName": "firefox"}}`)))
+	if err != nil {
+		err = errors.New("Send request, " + err.Error())
+		return
+	}
+	defer resp.Body.Close()
+	b, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		err = errors.New("Read response body, " + err.Error())
+		return
+	}
+
+	var message jsonwire.NewSession
+	err = json.Unmarshal(b, &message)
+	if err != nil {
+		err = errors.New("Unmarshal json, " + err.Error() + ", given response body=[" + string(b) + "]")
+		return
+	}
+	switch {
+	case message.SessionId != "":
+		sessionID = message.SessionId
+	case message.Value.SessionId != "":
+		sessionID = message.Value.SessionId
+	default:
+		err = errors.New("Field`s SessionId is empty")
+		return
+	}
+	if resp.StatusCode != http.StatusOK {
+		err = errors.New("Expected status code 200, actual: " + strconv.Itoa(resp.StatusCode) + ", given response body=[" + string(b) + "]")
+		return
+	}
+	return sessionID, nil
+}
+
+func sendAnyRequest(sessionID string) (err error) {
+	resp, err := http.Get(*hubUrl + "/wd/hub/session/" + sessionID + "/url")
+	if err != nil {
+		err = errors.New("Send request, " + err.Error())
+		return
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		b, _ := ioutil.ReadAll(resp.Body)
+		err = errors.New("Expected status code 200, actual: " + strconv.Itoa(resp.StatusCode) + ", given response body=[" + string(b) + "]")
+		return
+	}
+	return
+}
+
+func closeSession(sessionID string) (err error) {
+	req, err := http.NewRequest(http.MethodDelete, *hubUrl+"/wd/hub/session/"+sessionID, nil)
+	if err != nil {
+		err = errors.New("Create request, " + err.Error())
+		return
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		err = errors.New("Send request, " + err.Error())
+		return
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		b, _ := ioutil.ReadAll(resp.Body)
+		err = errors.New("Expected status code 200, actual: " + strconv.Itoa(resp.StatusCode) + ", given response body=[" + string(b) + "]")
+		return
+	}
+	return
+}