WIP: Improve tamper e2e test.

Co-authored-by: iknite <[email protected]>

gossip/auditor/auditor.go

@@ -76,6 +76,9 @@ func (t *MembershipTask) getSnapshot(version uint64) (*protocol.SignedSnapshot,
 		return nil, fmt.Errorf("Error getting snapshot from the store: %v", err)
 	}
 	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("Error getting snapshot from the store. Status: %d", resp.StatusCode)
+	}
 	buf, err := ioutil.ReadAll(resp.Body)
 	var s protocol.SignedSnapshot
 	err = s.Decode(buf)

tests/e2e/setup.go

@@ -60,7 +60,7 @@ func merge(list ...scope.TestF) scope.TestF {
 	}
 }
 
-func newAgent(id int, name string, role member.Type, p gossip.Processor, t *testing.T) {
+func newAgent(id int, name string, role member.Type, p gossip.Processor, t *testing.T) *gossip.Agent {
 	agentConf := gossip.DefaultConfig()
 	agentConf.NodeName = fmt.Sprintf("%s%d", name, id)
 
@@ -78,14 +78,16 @@ func newAgent(id int, name string, role member.Type, p gossip.Processor, t *test
 	agentConf.AlertsUrls = []string{PubUrl}
 	agentConf.Role = role
 
-	_, err := gossip.NewAgent(agentConf, []gossip.Processor{p})
+	agent, err := gossip.NewAgent(agentConf, []gossip.Processor{p})
 	if err != nil {
 		t.Fatalf("Failed to start the QED %s: %v", name, err)
 	}
+	return agent
 }
 
 func setupAuditor(id int, t *testing.T) (scope.TestF, scope.TestF) {
 	var au *auditor.Auditor
+	var agent *gossip.Agent
 	var err error
 
 	before := func(t *testing.T) {
@@ -99,12 +101,13 @@ func setupAuditor(id int, t *testing.T) (scope.TestF, scope.TestF) {
 			t.Fatalf("Unable to create a new auditor: %v", err)
 		}
 
-		newAgent(id, "auditor", member.Auditor, au, t)
+		agent = newAgent(id, "auditor", member.Auditor, au, t)
 	}
 
 	after := func(t *testing.T) {
 		if au != nil {
 			au.Shutdown()
+			fmt.Println(">>>>>>>>>>>>>", agent.Shutdown())
 		} else {
 			t.Fatalf("Unable to shutdown the auditor!")
 		}
@@ -114,6 +117,7 @@ func setupAuditor(id int, t *testing.T) (scope.TestF, scope.TestF) {
 
 func setupMonitor(id int, t *testing.T) (scope.TestF, scope.TestF) {
 	var mn *monitor.Monitor
+	var agent *gossip.Agent
 	var err error
 
 	before := func(t *testing.T) {
@@ -126,12 +130,13 @@ func setupMonitor(id int, t *testing.T) (scope.TestF, scope.TestF) {
 			t.Fatalf("Unable to create a new monitor: %v", err)
 		}
 
-		newAgent(id, "monitor", member.Monitor, mn, t)
+		agent = newAgent(id, "monitor", member.Monitor, mn, t)
 	}
 
 	after := func(t *testing.T) {
 		if mn != nil {
 			mn.Shutdown()
+			agent.Shutdown()
 		} else {
 			t.Fatalf("Unable to shutdown the monitor!")
 		}
@@ -141,6 +146,7 @@ func setupMonitor(id int, t *testing.T) (scope.TestF, scope.TestF) {
 
 func setupPublisher(id int, t *testing.T) (scope.TestF, scope.TestF) {
 	var pu *publisher.Publisher
+	var agent *gossip.Agent
 	var err error
 
 	before := func(t *testing.T) {
@@ -152,12 +158,13 @@ func setupPublisher(id int, t *testing.T) (scope.TestF, scope.TestF) {
 			t.Fatalf("Unable to create a new publisher: %v", err)
 		}
 
-		newAgent(id, "publisher", member.Publisher, pu, t)
+		agent = newAgent(id, "publisher", member.Publisher, pu, t)
 	}
 
 	after := func(t *testing.T) {
 		if pu != nil {
 			pu.Shutdown()
+			agent.Shutdown()
 		} else {
 			t.Fatalf("Unable to shutdown the publisher!")
 		}
@@ -168,10 +175,8 @@ func setupPublisher(id int, t *testing.T) (scope.TestF, scope.TestF) {
 func setupStore(t *testing.T) (scope.TestF, scope.TestF) {
 	var s *Service
 	before := func(t *testing.T) {
-		go (func() {
-			s = NewService()
-			s.Start()
-		})()
+		s = NewService()
+		s.Start()
 	}
 
 	after := func(t *testing.T) {

tests/e2e/tamper_test.go

@@ -16,35 +16,104 @@
 package e2e
 
 import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
 	"testing"
+	"time"
 
+	"github.com/bbva/qed/hashing"
+	"github.com/bbva/qed/protocol"
 	"github.com/bbva/qed/testutils/rand"
 	"github.com/bbva/qed/testutils/scope"
 	assert "github.com/stretchr/testify/require"
 )
 
+func getSnapshot(version uint64) (*protocol.SignedSnapshot, error) {
+	resp, err := http.Get(fmt.Sprintf("%s/snapshot?v=%d", PubUrl, version))
+	if err != nil {
+		return nil, fmt.Errorf("Error getting snapshot from the store: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("Error getting snapshot from the store. Status: %d", resp.StatusCode)
+	}
+
+	buf, _ := ioutil.ReadAll(resp.Body)
+	s := &protocol.SignedSnapshot{}
+	err = s.Decode(buf)
+	if err != nil {
+		return nil, fmt.Errorf("Error decoding signed snapshot %d codec", version)
+	}
+	return s, nil
+}
+
 func TestTamper(t *testing.T) {
 	bStore, aStore := setupStore(t)
 	bServer, aServer := setupServer(0, "", t)
-	bAuditor, aAuditor := setupAuditor(0, t)
-	bMonitor, aMonitor := setupMonitor(0, t)
+	// bAuditor, aAuditor := setupAuditor(0, t)
+	// bMonitor, aMonitor := setupMonitor(0, t)
 	bPublisher, aPublisher := setupPublisher(0, t)
 
 	scenario, let := scope.Scope(t,
-		merge(bStore, bServer, bAuditor, bMonitor, bPublisher),
-		merge(aStore, aServer, aAuditor, aMonitor, aPublisher),
+		merge(bStore, bServer /*bAuditor, bMonitor*/, bPublisher),
+		merge( /*aAuditor, aMonitor,*/ aPublisher, aServer, aStore),
 	)
 
 	client := getClient(0)
 
 	event := rand.RandomString(10)
 
+	scenario("Add one event and get its membership proof", func() {
+		var snapshot *protocol.Snapshot
+		var err error
+
+		let("Add event", func(t *testing.T) {
+			snapshot, err = client.Add(event)
+			assert.NoError(t, err)
+
+			assert.Equal(t, snapshot.EventDigest, hashing.NewSha256Hasher().Do([]byte(event)),
+				"The snapshot's event doesn't match: expected %s, actual %s", event, snapshot.EventDigest)
+			assert.False(t, snapshot.Version < 0, "The snapshot's version must be greater or equal to 0")
+			assert.False(t, len(snapshot.HyperDigest) == 0, "The snapshot's hyperDigest cannot be empty")
+			assert.False(t, len(snapshot.HistoryDigest) == 0, "The snapshot's hyperDigest cannot be empt")
+		})
+
+		let("Get membership proof for first inserted event", func(t *testing.T) {
+			result, err := client.Membership([]byte(event), snapshot.Version)
+			assert.NoError(t, err)
+
+			assert.True(t, result.Exists, "The queried key should be a member")
+			assert.Equal(t, result.QueryVersion, snapshot.Version,
+				"The query version doest't match the queried one: expected %d, actual %d", snapshot.Version, result.QueryVersion)
+			assert.Equal(t, result.ActualVersion, snapshot.Version,
+				"The actual version should match the queried one: expected %d, actual %d", snapshot.Version, result.ActualVersion)
+			assert.Equal(t, result.CurrentVersion, snapshot.Version,
+				"The current version should match the queried one: expected %d, actual %d", snapshot.Version, result.CurrentVersion)
+			assert.Equal(t, []byte(event), result.Key,
+				"The returned event doesn't math the original one: expected %s, actual %s", event, result.Key)
+			assert.False(t, len(result.KeyDigest) == 0, "The key digest cannot be empty")
+			assert.False(t, len(result.Hyper) == 0, "The hyper proof cannot be empty")
+			assert.False(t, result.ActualVersion > 0 && len(result.History) == 0,
+				"The history proof cannot be empty when version is greater than 0")
+
+		})
+	})
+
 	scenario("S", func() {
+		var snapshot *protocol.Snapshot
 		var err error
 
 		let("Add event", func(t *testing.T) {
-			_, err = client.Add(event)
+			snapshot, err = client.Add(event)
 			assert.NoError(t, err)
+			time.Sleep(100 * time.Second)
+			ss, err := getSnapshot(0)
+			if err != nil {
+				fmt.Println("Error: ", err)
+			}
+			assert.Equal(t, snapshot, ss.Snapshot)
 		})
 	})
+
 }

tests/e2e/test_service.go

@@ -14,6 +14,7 @@
 package e2e
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -154,7 +155,7 @@ func (s *Service) getSnapshotHandler() func(http.ResponseWriter, *http.Request)
 			}
 			b, ok := s.snaps.Get(uint64(version))
 			if !ok {
-				http.Error(w, fmt.Sprintf("Version not found: %v", version), http.StatusMethodNotAllowed)
+				http.Error(w, fmt.Sprintf("Version not found: %v", version), http.StatusUnprocessableEntity)
 				return
 			}
 			buf, err := b.Encode()
@@ -167,6 +168,7 @@ func (s *Service) getSnapshotHandler() func(http.ResponseWriter, *http.Request)
 		http.Error(w, "Invalid request method", http.StatusMethodNotAllowed)
 	}
 }
+
 func (s *Service) alertHandler() func(http.ResponseWriter, *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		atomic.AddUint64(&s.stats.count[RPS], 1)
@@ -209,6 +211,7 @@ func NewService() *Service {
 	snaps.d = make(map[uint64]*protocol.SignedSnapshot, 0)
 	stats.batch = make(map[string][]int, 0)
 	alerts.d = make([]string, 0)
+
 	return &Service{
 		snaps:  &snaps,
 		alerts: &alerts,
@@ -218,6 +221,13 @@ func NewService() *Service {
 }
 
 func (s *Service) Start() {
+	router := http.NewServeMux()
+	router.HandleFunc("/stat", s.statHandler())
+	router.HandleFunc("/batch", s.postBatchHandler())
+	router.HandleFunc("/snapshot", s.getSnapshotHandler())
+	router.HandleFunc("/alert", s.alertHandler())
+
+	httpServer := &http.Server{Addr: "127.0.0.1:8888", Handler: router}
 	go func() {
 		ticker := time.NewTicker(1 * time.Second)
 		for {
@@ -228,16 +238,18 @@ func (s *Service) Start() {
 				fmt.Println("Counters ", s.stats.count)
 				atomic.StoreUint64(&s.stats.count[RPS], 0)
 			case <-s.quitCh:
+				fmt.Println("\nShutting down the server...")
+				httpServer.Shutdown(context.Background())
 				return
 			}
 		}
 	}()
 
-	http.HandleFunc("/stat", s.statHandler())
-	http.HandleFunc("/batch", s.postBatchHandler())
-	http.HandleFunc("/snapshot", s.getSnapshotHandler())
-	http.HandleFunc("/alert", s.alertHandler())
-	log.Fatal(http.ListenAndServe("127.0.0.1:8888", nil))
+	go (func() {
+		if err := httpServer.ListenAndServe(); err != http.ErrServerClosed {
+			log.Fatal(err)
+		}
+	})()
 }
 
 func (s *Service) Shutdown() {