Repository: allegro/marathon-consul
Pull request #244 (2017-06-19 15:05:49)

Add tests to streamer

coveralls: 
[![Coverage Status](https://:/builds/12044812/badge)](https://:/builds/12044812)

Coverage increased (+2.6%) to 89.38% when pulling **29601de5f9ab4a25a0ba9d41899fc5de2109851b on add_tests_to_streamer** into **61512988126183cd520b459aafa5f4280b3cb18f on master**.

coveralls: 
[![Coverage Status](https://:/builds/12045774/badge)](https://:/builds/12045774)

Coverage increased (+2.6%) to 89.374% when pulling **230508c31f35e0a89da7a465f5f55284d690d58c on add_tests_to_streamer** into **9732f5665118e3aa90862739d359cc5fd5ac0d07 on master**.

coveralls: 
[![Coverage Status](https://:/builds/12046361/badge)](https://:/builds/12046361)

Coverage increased (+2.6%) to 89.374% when pulling **230508c31f35e0a89da7a465f5f55284d690d58c on add_tests_to_streamer** into **9732f5665118e3aa90862739d359cc5fd5ac0d07 on master**.


Patch:
diff --git a/marathon/streamer.go b/marathon/streamer.go
index 0353b2b..a5d96e2 100644
--- a/marathon/streamer.go
+++ b/marathon/streamer.go
@@ -3,6 +3,7 @@ package marathon
 import (
 	"bufio"
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"time"
@@ -22,19 +23,15 @@ type Streamer struct {
 	noRecover    bool
 }
 
-func (s *Streamer) stop() {
-	s.cancel()
-}
 func (s *Streamer) Stop() {
+	s.cancel()
 	s.noRecover = true
-	s.stop()
 }
 
 func (s *Streamer) Start() error {
 	req, err := http.NewRequest("GET", s.subURL, nil)
 	if err != nil {
-		log.Fatal("Unable to create Streamer request")
-		return nil
+		return fmt.Errorf("Unable to create request: %s", err)
 	}
 	req.SetBasicAuth(s.username, s.password)
 	req.Header.Set("Accept", "text/event-stream")
@@ -44,14 +41,10 @@ func (s *Streamer) Start() error {
 	res, err := s.client.Do(req)
 	if err != nil {
 		s.cancel()
-		return err
+		return fmt.Errorf("Subscription request errored: %s", err)
 	}
 	if res.StatusCode != http.StatusOK {
-		log.WithFields(log.Fields{
-			"Host":   req.Host,
-			"URI":    req.URL.RequestURI(),
-			"Method": "GET",
-		}).Errorf("Got status code : %d", res.StatusCode)
+		return fmt.Errorf("Event stream not connected: Expected %d but got %d", http.StatusOK, res.StatusCode)
 	}
 	log.WithFields(log.Fields{
 		"Host":   req.Host,
@@ -65,9 +58,9 @@ func (s *Streamer) Start() error {
 
 func (s *Streamer) Recover() error {
 	if s.noRecover {
-		return fmt.Errorf("Streamer is not recoverable")
+		return errors.New("Streamer is not recoverable")
 	}
-	s.stop()
+	s.cancel()
 
 	err := s.Start()
 	i := 0
@@ -76,5 +69,8 @@ func (s *Streamer) Recover() error {
 		time.Sleep(seconds * time.Second)
 		i++
 	}
-	return err
+	if err != nil {
+		return fmt.Errorf("Cannot recover Streamer: %s", err)
+	}
+	return nil
 }


Test patch:
diff --git a/marathon/streamer_test.go b/marathon/streamer_test.go
new file mode 100644
index 0000000..0cb8806
--- /dev/null
+++ b/marathon/streamer_test.go
@@ -0,0 +1,97 @@
+package marathon
+
+import (
+	"testing"
+
+	"net/http"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStreamer_RecoverShouldReturnErrorWhenCantStart(t *testing.T) {
+	server, transport := stubServer("/v2/events", "")
+	defer server.Close()
+	client := &http.Client{Transport: transport}
+	s := Streamer{client: client}
+	s.Start()
+
+	err := s.Recover()
+
+	assert.EqualError(t, err, "Cannot recover Streamer: Subscription request errored: Get : unsupported protocol scheme \"\"")
+}
+
+func TestStreamer_StartShouldReturnErrorOnInvalidUrl(t *testing.T) {
+	s := Streamer{client: http.DefaultClient}
+
+	err := s.Start()
+
+	assert.EqualError(t, err, "Subscription request errored: Get : unsupported protocol scheme \"\"")
+}
+
+func TestStreamer_StartShouldReturnErrorOnNon200Response(t *testing.T) {
+	server, transport := stubServer("/v2/events", "")
+	defer server.Close()
+	client := &http.Client{Transport: transport}
+	s := Streamer{client: client, subURL: "http://marathon/invalid/path"}
+
+	err := s.Start()
+
+	assert.EqualError(t, err, "Event stream not connected: Expected 200 but got 404")
+}
+
+func TestStreamer_StartShouldReturnNoErrorIfSuccessfulConnects(t *testing.T) {
+	server, transport := stubServer("/v2/events", "")
+	defer server.Close()
+	client := &http.Client{Transport: transport}
+	s := Streamer{client: client, subURL: "http://marathon/v2/events"}
+
+	err := s.Start()
+
+	assert.NoError(t, err)
+}
+
+func TestStreamer_RecoverShouldReturnNoErrorIfSuccessfulConnects(t *testing.T) {
+	server, transport := stubServer("/v2/events", "")
+	defer server.Close()
+	client := &http.Client{Transport: transport}
+	s := Streamer{client: client, subURL: "http://marathon/v2/events"}
+	s.Start()
+
+	err := s.Recover()
+
+	assert.NoError(t, err)
+}
+
+func TestStreamer_StopShouldPreventRecovery(t *testing.T) {
+	server, transport := stubServer("/v2/events", "")
+	defer server.Close()
+	client := &http.Client{Transport: transport}
+	s := Streamer{client: client, subURL: "http://marathon/v2/events"}
+
+	s.Start()
+	s.Stop()
+	err := s.Recover()
+
+	assert.Error(t, err, "Streamer is not recoverable")
+}
+
+func TestStreamer_StopShouldCancelRequest(t *testing.T) {
+	ready := make(chan bool)
+	wait := make(chan bool)
+	server, transport := mockServer(func(w http.ResponseWriter, r *http.Request) {
+		ready <- true
+		<-wait
+	})
+	defer server.Close()
+	client := &http.Client{Transport: transport}
+	s := Streamer{client: client, subURL: "http://marathon/v2/events"}
+
+	go func() {
+		err := s.Start()
+		assert.Error(t, err, "Subscription request errored: Get http://marathon/v2/events: net/http: request canceled")
+	}()
+
+	<-ready
+	s.Stop()
+	wait <- false
+}