func TestRouter_SimpleMessageSending(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	a := assert.New(t)

	// Given a Router with route
	router, r := aRouterRoute(chanSize)
	msMock := NewMockMessageStore(ctrl)
	router.messageStore = msMock

	id, ts := uint64(2), time.Now().Unix()
	msMock.EXPECT().
		StoreMessage(gomock.Any(), gomock.Any()).
		Do(func(m *protocol.Message, nodeID uint8) (int, error) {
			m.ID = id
			m.Time = ts
			m.NodeID = nodeID
			return len(m.Bytes()), nil
		})

	// when i send a message to the route
	router.HandleMessage(&protocol.Message{Path: r.Path, Body: aTestByteMessage})

	// then I can receive it a short time later
	assertChannelContainsMessage(a, r.MessagesChannel(), aTestByteMessage)
}

func TestHealthUp(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()
	defer testutil.ResetDefaultRegistryHealthCheck()
	a := assert.New(t)

	// given:
	service, _, _, _ := aMockedServiceWithMockedRouterStandalone()
	service = service.HealthEndpoint("/health_url")
	a.Equal(2, len(service.ModulesSortedByStartOrder()))

	// when starting the service
	defer service.Stop()
	service.Start()
	time.Sleep(time.Millisecond * 10)

	// and when I call the health URL
	url := fmt.Sprintf("http://%s/health_url", service.WebServer().GetAddr())
	result, err := http.Get(url)

	// then I get status 200 and JSON: {}
	a.NoError(err)
	body, err := ioutil.ReadAll(result.Body)
	a.NoError(err)
	a.Equal("{}", string(body))
}

func TestServerHTTP(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	// given:  a rest api with a message sink
	routerMock := NewMockRouter(ctrl)
	api := NewRestMessageAPI(routerMock, "/api")

	url, _ := url.Parse("http://localhost/api/message/my/topic?userId=marvin&messageId=42")

	// and a http context
	req := &http.Request{
		Method: http.MethodPost,
		URL:    url,
		Body:   ioutil.NopCloser(bytes.NewReader(testBytes)),
		Header: http.Header{},
	}
	w := &httptest.ResponseRecorder{}

	// then i expect
	routerMock.EXPECT().HandleMessage(gomock.Any()).Do(func(msg *protocol.Message) {
		a.Equal(testBytes, msg.Body)
		a.Equal("{}", msg.HeaderJSON)
		a.Equal("/my/topic", string(msg.Path))
		a.True(len(msg.ApplicationID) > 0)
		a.Nil(msg.Filters)
		a.Equal("marvin", msg.UserID)
	})

	// when: I POST a message
	api.ServeHTTP(w, req)

}

func TestConnector_PostSubscriptionNoMocks(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	recorder := httptest.NewRecorder()
	conn, mocks := getTestConnector(t, Config{
		Name:       "test",
		Schema:     "test",
		Prefix:     "/connector/",
		URLPattern: "/{device_token}/{user_id}/{topic:.*}",
	}, false, false)

	entriesC := make(chan [2]string)
	mocks.kvstore.EXPECT().Iterate(gomock.Eq("test"), gomock.Eq("")).Return(entriesC)
	close(entriesC)

	mocks.kvstore.EXPECT().Put(gomock.Eq("test"), gomock.Eq(GenerateKey("/topic1", map[string]string{
		"device_token": "device1",
		"user_id":      "user1",
	})), gomock.Any())

	mocks.router.EXPECT().Subscribe(gomock.Any())

	err := conn.Start()
	a.NoError(err)
	defer conn.Stop()

	req, err := http.NewRequest(http.MethodPost, "/connector/device1/user1/topic1", strings.NewReader(""))
	a.NoError(err)
	conn.ServeHTTP(recorder, req)
	a.Equal(`{"subscribed":"/topic1"}`, recorder.Body.String())
	time.Sleep(100 * time.Millisecond)
}

func TestGCMConnector_GetErrorMessageFromGcm(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	// defer testutil.EnableDebugForMethod()()

	a := assert.New(t)
	routerMock := NewMockRouter(ctrl)
	routerMock.EXPECT().Subscribe(gomock.Any()).Do(func(route *server.Route) {
		a.Equal("/gcm/broadcast", string(route.Path))
		a.Equal("gcm_connector", route.UserID)
		a.Equal("gcm_connector", route.ApplicationID)
	})

	// expect the route unsubscribed from removeSubscription
	routerMock.EXPECT().Unsubscribe(gomock.Any()).Do(func(route *server.Route) {
		a.Equal("/path", string(route.Path))
		a.Equal("id", route.ApplicationID)
	})

	// expect the route subscribe with the new canonicalId from replaceSubscriptionWithCanonicalID
	routerMock.EXPECT().Subscribe(gomock.Any()).Do(func(route *server.Route) {
		a.Equal("/path", string(route.Path))
		a.Equal("marvin", route.UserID)
		a.Equal("gcmCanonicalID", route.ApplicationID)
	})

	kvStore := store.NewMemoryKVStore()
	routerMock.EXPECT().KVStore().Return(kvStore, nil)

	gcm, err := NewGCMConnector(routerMock, "/gcm/", "testApi", 1)
	a.Nil(err)

	err = gcm.Start()
	a.Nil(err)

	done := make(chan bool, 1)
	mockSender := testutil.CreateGcmSender(
		testutil.CreateRoundTripperWithJsonResponse(http.StatusOK, testutil.ErrorResponseMessageJSON, done))
	gcm.Sender = mockSender

	// put a dummy gcm message with minimum information
	msg := &server.MessageForRoute{
		Message: &protocol.Message{
			ID:   uint64(4),
			Body: []byte("{id:id}"),
			Time: 1405544146,
			Path: "/gcm/marvin/gcm124/subscribe/stuff"},
		Route: &server.Route{
			ApplicationID: "id",
			Path:          "/path",
			UserID:        "marvin"}}

	gcm.routerC <- msg
	// expect that the Http Server gives us a malformed message
	<-done
	//wait before closing the gcm connector
	time.Sleep(50 * time.Millisecond)
	err = gcm.Stop()
	a.NoError(err)
}

func TestConnector_GetList_And_Getters(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	recorder := httptest.NewRecorder()
	conn, mocks := getTestConnector(t, Config{
		Name:       "test",
		Schema:     "test",
		Prefix:     "/connector/",
		URLPattern: "/{device_token}/{user_id}/{topic:.*}",
	}, true, false)

	req, err := http.NewRequest(http.MethodGet, "/connector/", strings.NewReader(""))
	a.NoError(err)

	conn.ServeHTTP(recorder, req)
	expectedJSON := `{"error":"Missing filters"}`
	a.JSONEq(expectedJSON, recorder.Body.String())
	a.Equal(http.StatusBadRequest, recorder.Code)

	a.Equal("/connector/", conn.GetPrefix())
	a.Equal(mocks.manager, conn.Manager())
	a.Equal(nil, conn.ResponseHandler())
}

func TestConnector_GetListWithFilters(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	recorder := httptest.NewRecorder()
	conn, mocks := getTestConnector(t, Config{
		Name:       "test",
		Schema:     "test",
		Prefix:     "/connector/",
		URLPattern: "/{device_token}/{user_id}/{topic:.*}",
	}, true, false)

	mocks.manager.EXPECT().Filter(gomock.Eq(map[string]string{
		"filter1": "value1",
		"filter2": "value2",
	})).Return([]Subscriber{})

	req, err := http.NewRequest(
		http.MethodGet,
		"/connector/?filter1=value1&filter2=value2",
		strings.NewReader(""))
	a.NoError(err)

	conn.ServeHTTP(recorder, req)
}

func TestConn_HandleResponse(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()
	a := assert.New(t)

	//given
	c, mKVS := newAPNSConnector(t)

	mSubscriber := NewMockSubscriber(testutil.MockCtrl)
	mSubscriber.EXPECT().SetLastID(gomock.Any())
	mSubscriber.EXPECT().Key().Return("key").AnyTimes()
	mSubscriber.EXPECT().Encode().Return([]byte("{}"), nil).AnyTimes()
	mKVS.EXPECT().Put(schema, "key", []byte("{}")).Times(2)

	c.Manager().Add(mSubscriber)

	message := &protocol.Message{
		ID: 42,
	}
	mRequest := NewMockRequest(testutil.MockCtrl)
	mRequest.EXPECT().Message().Return(message).AnyTimes()
	mRequest.EXPECT().Subscriber().Return(mSubscriber).AnyTimes()

	response := &apns2.Response{
		ApnsID:     "id-life",
		StatusCode: 200,
	}

	//when
	err := c.HandleResponse(mRequest, response, nil, nil)

	//then
	a.NoError(err)
}

func TestConnector_DeleteSubscription(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	recorder := httptest.NewRecorder()
	conn, mocks := getTestConnector(t, Config{
		Name:       "test",
		Schema:     "test",
		Prefix:     "/connector/",
		URLPattern: "/{device_token}/{user_id}/{topic:.*}",
	}, true, false)

	subscriber := NewMockSubscriber(testutil.MockCtrl)
	mocks.manager.EXPECT().Find(gomock.Eq(GenerateKey("/topic1", map[string]string{
		"device_token": "device1",
		"user_id":      "user1",
	}))).Return(subscriber)
	mocks.manager.EXPECT().Remove(subscriber).Return(nil)

	req, err := http.NewRequest(http.MethodDelete, "/connector/device1/user1/topic1", strings.NewReader(""))
	a.NoError(err)
	conn.ServeHTTP(recorder, req)
	a.Equal(`{"unsubscribed":"/topic1"}`, recorder.Body.String())
	time.Sleep(200 * time.Millisecond)
}

func TestSendUnSubscribeMessage(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()

	// given a client
	c := New("url", "origin", 1, true)

	// when expects a message
	connMock := NewMockWSConnection(ctrl)
	connMock.EXPECT().WriteMessage(websocket.BinaryMessage, []byte("- /foo"))
	connMock.EXPECT().
		ReadMessage().
		Return(websocket.BinaryMessage, []byte(aNormalMessage), nil).
		Do(func() {
			time.Sleep(time.Millisecond * 50)
		}).
		AnyTimes()
	c.SetWSConnectionFactory(MockConnectionFactory(connMock))

	c.Start()
	c.Unsubscribe("/foo")

	// stop client after 200ms
	time.AfterFunc(time.Millisecond*200, func() { c.Close() })
}

func TestSendAMessage(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	//	a := assert.New(t)

	// given a client
	c := New("url", "origin", 1, true)

	// when expects a message
	connMock := NewMockWSConnection(ctrl)
	connMock.EXPECT().WriteMessage(websocket.BinaryMessage, []byte("> /foo\n{}\nTest"))
	connMock.EXPECT().
		ReadMessage().
		Return(websocket.BinaryMessage, []byte(aNormalMessage), nil).
		Do(func() {
			time.Sleep(time.Millisecond * 50)
		}).
		AnyTimes()
	c.SetWSConnectionFactory(MockConnectionFactory(connMock))

	c.Start()
	// then the expectation is meet by sending it
	c.Send("/foo", "Test", "{}")
	// stop client after 200ms
	time.AfterFunc(time.Millisecond*200, func() { c.Close() })
}

func TestStopableClient(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	a := assert.New(t)

	// given a client
	c := New("url", "origin", 1, true)

	// with a closeable connection
	connMock := NewMockWSConnection(ctrl)
	close := make(chan bool, 1)
	connMock.EXPECT().ReadMessage().
		Do(func() { <-close }).
		Return(0, []byte{}, fmt.Errorf("expected close error"))

	connMock.EXPECT().Close().Do(func() {
		close <- true
	})

	c.SetWSConnectionFactory(MockConnectionFactory(connMock))

	// when we start
	err := c.Start()

	// than we are connected
	a.NoError(err)
	a.True(c.IsConnected())

	// when we clode
	c.Close()
	time.Sleep(time.Millisecond * 1)

	// than the client returns
	a.False(c.IsConnected())
}

func Test_BadCommands(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	badRequests := []string{"XXXX", "", ">", ">/foo", "+", "-", "send /foo"}
	wsconn, routerMock, messageStore := createDefaultMocks(badRequests)

	counter := 0

	var wg sync.WaitGroup
	wg.Add(len(badRequests))

	wsconn.EXPECT().Send(gomock.Any()).Do(func(data []byte) error {
		if strings.HasPrefix(string(data), "#connected") {
			return nil
		}
		if strings.HasPrefix(string(data), "!error-bad-request") {
			counter++
		} else {
			t.Logf("expected bad-request, but got: %v", string(data))
		}

		wg.Done()
		return nil
	}).AnyTimes()

	runNewWebSocket(wsconn, routerMock, messageStore, nil)

	wg.Wait()
	assert.Equal(t, len(badRequests), counter, "expected number of bad requests does not match")
}

func TestHealthDown(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	defer testutil.ResetDefaultRegistryHealthCheck()
	a := assert.New(t)

	// given:
	service, _, _, _ := aMockedServiceWithMockedRouterStandalone()
	service = service.HealthEndpoint("/health_url")
	mockChecker := NewMockChecker(ctrl)
	mockChecker.EXPECT().Check().Return(errors.New("sick")).AnyTimes()

	// when starting the service with a short frequency
	defer service.Stop()
	service.healthFrequency = time.Millisecond * 3
	service.RegisterModules(0, 0, mockChecker)
	a.Equal(3, len(service.ModulesSortedByStartOrder()))

	service.Start()
	time.Sleep(time.Millisecond * 10)

	// and when I can call the health URL
	url := fmt.Sprintf("http://%s/health_url", service.WebServer().GetAddr())
	result, err := http.Get(url)
	// then I receive status 503 and a JSON error message
	a.NoError(err)
	a.Equal(503, result.StatusCode)
	body, err := ioutil.ReadAll(result.Body)
	a.NoError(err)
	a.Equal("{\"*service.MockChecker\":\"sick\"}", string(body))
}

func TestServeHTTPSuccess(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	// given: a rest api with a message sink
	routerMock := NewMockRouter(ctrl)

	kvStore := store.NewMemoryKVStore()
	routerMock.EXPECT().KVStore().Return(kvStore, nil)

	routerMock.EXPECT().Subscribe(gomock.Any()).Do(func(route *server.Route) {
		a.Equal("/notifications", string(route.Path))
		a.Equal("marvin", route.UserID)
		a.Equal("gcmId123", route.ApplicationID)
	})

	gcm, err := NewGCMConnector(routerMock, "/gcm/", "testApi", 1)
	a.Nil(err)

	url, _ := url.Parse("http://localhost/gcm/marvin/gcmId123/subscribe/notifications")
	// and a http context
	req := &http.Request{URL: url, Method: "POST"}
	w := httptest.NewRecorder()

	// when: I POST a message
	gcm.ServeHTTP(w, req)

	// then
	a.Equal("registered: /notifications\n", string(w.Body.Bytes()))
}

func TestReadLastID(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()

	defer testutil.EnableDebugForMethod()()
	a := assert.New(t)

	mockSmsSender := NewMockSender(ctrl)
	kvStore := kvstore.NewMemoryKVStore()

	a.NotNil(kvStore)
	routerMock := NewMockRouter(testutil.MockCtrl)
	routerMock.EXPECT().KVStore().AnyTimes().Return(kvStore, nil)
	msgStore := dummystore.New(kvStore)
	routerMock.EXPECT().MessageStore().AnyTimes().Return(msgStore, nil)

	topic := "/sms"
	worker := 1
	config := Config{
		Workers:  &worker,
		SMSTopic: &topic,
		Name:     "test_gateway",
		Schema:   SMSSchema,
	}

	gw, err := New(routerMock, mockSmsSender, config)
	a.NoError(err)

	gw.SetLastSentID(uint64(10))

	gw.ReadLastID()

	a.Equal(uint64(10), gw.LastIDSent)
}

func TestGCMConnector_Check(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()
	testutil.EnableDebugForMethod()

	assert := assert.New(t)
	routerMock := NewMockRouter(ctrl)
	routerMock.EXPECT().Subscribe(gomock.Any()).Do(func(route *server.Route) {
		assert.Equal("/gcm/broadcast", string(route.Path))
		assert.Equal("gcm_connector", route.UserID)
		assert.Equal("gcm_connector", route.ApplicationID)
	})

	kvStore := store.NewMemoryKVStore()
	routerMock.EXPECT().KVStore().Return(kvStore, nil)

	gcm, err := NewGCMConnector(routerMock, "/gcm/", "testApi", 1)
	assert.Nil(err)

	err = gcm.Start()
	assert.Nil(err)

	done := make(chan bool, 1)
	mockSender := testutil.CreateGcmSender(testutil.CreateRoundTripperWithJsonResponse(http.StatusOK, testutil.CorrectGcmResponseMessageJSON, done))
	gcm.Sender = mockSender
	err = gcm.Check()
	fmt.Println(err)

	done2 := make(chan bool, 1)
	mockSender2 := testutil.CreateGcmSender(testutil.CreateRoundTripperWithJsonResponse(http.StatusUnauthorized, "", done2))
	gcm.Sender = mockSender2
	err = gcm.Check()
	fmt.Println(err)
}

func Test_AnIncomingMessageIsNotAllowed(t *testing.T) {
	ctrl, finish := testutil.NewMockCtrl(t)
	defer finish()

	wsconn, routerMock, _ := createDefaultMocks([]string{})

	tam := NewMockAccessManager(ctrl)
	tam.EXPECT().IsAllowed(auth.READ, "testuser", protocol.Path("/foo")).Return(false)
	handler := NewWebSocket(
		testWSHandler(routerMock, tam),
		wsconn,
		"testuser",
	)
	go func() {
		handler.Start()
	}()
	time.Sleep(time.Millisecond * 2)

	handler.sendChannel <- aTestMessage.Bytes()
	time.Sleep(time.Millisecond * 2)
	//nothing shall have been sent

	//now allow
	tam.EXPECT().IsAllowed(auth.READ, "testuser", protocol.Path("/foo")).Return(true)

	wsconn.EXPECT().Send(aTestMessage.Bytes())

	time.Sleep(time.Millisecond * 2)

	handler.sendChannel <- aTestMessage.Bytes()
	time.Sleep(time.Millisecond * 2)
}

func TestConnector_GetErrorMessageFromFCM(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)
	fcm, mocks := testFCM(t, true)

	err := fcm.Start()
	a.NoError(err)

	var route *router.Route

	mocks.router.EXPECT().Subscribe(gomock.Any()).Do(func(r *router.Route) (*router.Route, error) {
		a.Equal("/topic", string(r.Path))
		a.Equal("user01", r.Get("user_id"))
		a.Equal("device01", r.Get(deviceTokenKey))
		route = r
		return r, nil
	})

	// put a dummy FCM message with minimum information
	postSubscription(t, fcm, "user01", "device01", "topic")
	time.Sleep(100 * time.Millisecond)
	a.NoError(err)
	a.NotNil(route)

	// expect the route unsubscribed
	mocks.router.EXPECT().Unsubscribe(gomock.Any()).Do(func(route *router.Route) {
		a.Equal("/topic", string(route.Path))
		a.Equal("device01", route.Get(deviceTokenKey))
	})

	// expect the route subscribe with the new canonicalID from replaceSubscriptionWithCanonicalID
	mocks.router.EXPECT().Subscribe(gomock.Any()).Do(func(route *router.Route) {
		a.Equal("/topic", string(route.Path))
		a.Equal("user01", route.Get("user_id"))
		appid := route.Get(deviceTokenKey)
		a.Equal("fcmCanonicalID", appid)
	})
	// mocks.store.EXPECT().MaxMessageID(gomock.Any()).Return(uint64(4), nil)

	response := new(gcm.Response)
	err = json.Unmarshal([]byte(ErrorFCMResponse), response)
	a.NoError(err)
	mocks.gcmSender.EXPECT().Send(gomock.Any()).Return(response, nil)

	// send the message into the subscription route channel
	route.Deliver(&protocol.Message{
		ID:   uint64(4),
		Path: "/topic",
		Body: []byte("{id:id}"),
	})

	// wait before closing the FCM connector
	time.Sleep(100 * time.Millisecond)

	err = fcm.Stop()
	a.NoError(err)
}

func Test_Receiver_Fetch_Produces_Correct_Fetch_Requests(t *testing.T) {
	_, finish := testutil.NewMockCtrl(t)
	defer finish()

	a := assert.New(t)

	testcases := []struct {
		desc   string
		arg    string
		maxID  int
		expect store.FetchRequest
	}{
		{desc: "simple forward fetch",
			arg:    "/foo 0 20",
			maxID:  -1,
			expect: store.FetchRequest{Partition: "foo", Direction: 1, StartID: uint64(0), Count: 20},
		},
		{desc: "forward fetch without bounds",
			arg:    "/foo 0",
			maxID:  -1,
			expect: store.FetchRequest{Partition: "foo", Direction: 1, StartID: uint64(0), Count: math.MaxInt32},
		},
		{desc: "backward fetch to top",
			arg:    "/foo -20",
			maxID:  42,
			expect: store.FetchRequest{Partition: "foo", Direction: -1, StartID: uint64(42), Count: 20},
		},
		{desc: "backward fetch with count",
			arg:    "/foo -1 10",
			maxID:  42,
			expect: store.FetchRequest{Partition: "foo", Direction: -1, StartID: uint64(42), Count: 10},
		},
	}

	for _, test := range testcases {
		rec, _, _, messageStore, err := aMockedReceiver(test.arg)

		a.NotNil(rec)
		a.NoError(err, test.desc)

		if test.maxID != -1 {
			messageStore.EXPECT().MaxMessageID(test.expect.Partition).
				Return(uint64(test.maxID), nil)
		}

		done := make(chan bool)
		messageStore.EXPECT().Fetch(gomock.Any()).Do(func(r *store.FetchRequest) {
			a.Equal(test.expect.Partition, r.Partition, test.desc)
			a.Equal(test.expect.Direction, r.Direction, test.desc)
			a.Equal(test.expect.StartID, r.StartID, test.desc)
			a.Equal(test.expect.Count, r.Count, test.desc)
			done <- true
		})

		go rec.fetchOnlyLoop()
		testutil.ExpectDone(a, done)
		rec.Stop()
	}
}