Merge pull request #349 from sylwiaszunejko/ring-describer-testable

Move getting peers information to separate function

host_source.go

@@ -823,7 +823,6 @@ func (r *ringDescriber) getClusterPeerInfo(localHost *HostInfo) ([]*HostInfo, er
 		return nil, errNoControl
 	}
 
-	var peers []*HostInfo
 	iter := r.session.control.withConnHost(func(ch *connHost) *Iter {
 		return ch.conn.querySystemPeers(context.TODO(), localHost.version)
 	})
@@ -838,14 +837,20 @@ func (r *ringDescriber) getClusterPeerInfo(localHost *HostInfo) ([]*HostInfo, er
 		return nil, fmt.Errorf("unable to fetch peer host info: %s", err)
 	}
 
-	for _, row := range rows {
+	return getPeersFromQuerySystemPeers(rows, r.session.cfg.Port, r.session.cfg.translateAddressPort, r.session.logger)
+}
+
+func getPeersFromQuerySystemPeers(querySystemPeerRows []map[string]interface{}, port int, translateAddressPort func(addr net.IP, port int) (net.IP, int), logger StdLogger) ([]*HostInfo, error) {
+	var peers []*HostInfo
+
+	for _, row := range querySystemPeerRows {
 		// extract all available info about the peer
-		host, err := hostInfoFromMap(row, &HostInfo{port: r.session.cfg.Port}, r.session.cfg.translateAddressPort)
+		host, err := hostInfoFromMap(row, &HostInfo{port: port}, translateAddressPort)
 		if err != nil {
 			return nil, err
 		} else if !isValidPeer(host) {
 			// If it's not a valid peer
-			r.session.logger.Printf("Found invalid peer '%s' "+
+			logger.Printf("Found invalid peer '%s' "+
 				"Likely due to a gossip or snitch issue, this host will be ignored", host)
 			continue
 		} else if isZeroToken(host) {

host_source_test.go

@@ -364,3 +364,81 @@ func TestErrorBroadcaster_StopWithoutBroadcast(t *testing.T) {
 		t.Errorf(loadedVal.(error).Error())
 	}
 }
+
+func TestGetClusterPeerInfoZeroToken(t *testing.T) {
+	host_id1, _ := ParseUUID("b2035fd9-e0ca-4857-8c45-e63c00fb7c43")
+	host_id2, _ := ParseUUID("4b21ee4c-acea-4267-8e20-aaed5361a0dd")
+	host_id3, _ := ParseUUID("dfef4a22-b8d8-47e9-aee5-8c19d4b7a9e3")
+
+	schema_version1, _ := ParseUUID("af810386-a694-11ef-81fa-3aea73156247")
+
+	peersRows := []map[string]interface{}{
+		{
+			"data_center":     "datacenter1",
+			"host_id":         host_id1,
+			"peer":            "127.0.0.3",
+			"preferred_ip":    "127.0.0.3",
+			"rack":            "rack1",
+			"release_version": "3.0.8",
+			"rpc_address":     "127.0.0.3",
+			"schema_version":  schema_version1,
+			"tokens":          []string{"-1296227678594315580994457470329811265"},
+		},
+		{
+			"data_center":     "datacenter1",
+			"host_id":         host_id2,
+			"peer":            "127.0.0.2",
+			"preferred_ip":    "127.0.0.2",
+			"rack":            "rack1",
+			"release_version": "3.0.8",
+			"rpc_address":     "127.0.0.2",
+			"schema_version":  schema_version1,
+			"tokens":          []string{"-1129762924682054333"},
+		},
+		{
+			"data_center":     "datacenter2",
+			"host_id":         host_id3,
+			"peer":            "127.0.0.5",
+			"preferred_ip":    "127.0.0.5",
+			"rack":            "rack1",
+			"release_version": "3.0.8",
+			"rpc_address":     "127.0.0.5",
+			"schema_version":  schema_version1,
+			"tokens":          []string{},
+		},
+	}
+
+	translateAddressPort := func(addr net.IP, port int) (net.IP, int) {
+		return addr, port
+	}
+
+	var logger StdLogger
+	t.Run("OmitOneZeroTokenNode", func(t *testing.T) {
+		peers, err := getPeersFromQuerySystemPeers(
+			peersRows,
+			9042,
+			translateAddressPort,
+			logger,
+		)
+
+		if err != nil {
+			t.Fatalf("unable to get peers: %v", err)
+		}
+		assertEqual(t, "peers length", 2, len(peers))
+	})
+
+	t.Run("NoZeroTokenNodes", func(t *testing.T) {
+		peersRows[2]["tokens"] = []string{"-1129762924682054333"}
+		peers, err := getPeersFromQuerySystemPeers(
+			peersRows,
+			9042,
+			translateAddressPort,
+			logger,
+		)
+
+		if err != nil {
+			t.Fatalf("unable to get peers: %v", err)
+		}
+		assertEqual(t, "peers length", 3, len(peers))
+	})
+}