test: port peering test from sharness to Go

This is the slowest test in the sharness test suite, because it has
very long sleeps. It usually takes 2+ minutes to run.

This new impl runs all peering tests in about 20 seconds, since it
polls for conditions instead of sleeping, and runs the tests in
parallel.

This also has an additional test case for a peer that was never online
and then connects.

test/cli/harness/harness.go

@@ -11,6 +11,8 @@ import (
 
 	logging "github.com/ipfs/go-log/v2"
 	. "github.com/ipfs/kubo/test/cli/testutils"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/multiformats/go-multiaddr"
 )
 
 // Harness tracks state for a test, such as temp dirs and IFPS nodes, and cleans them up after the test.
@@ -188,3 +190,22 @@ func (h *Harness) Cleanup() {
 		log.Panicf("removing temp dir %s: %s", h.Dir, err)
 	}
 }
+
+// ExtractPeerID extracts a peer ID from the given multiaddr, and fatals if it does not contain a peer ID.
+func (h *Harness) ExtractPeerID(m multiaddr.Multiaddr) peer.ID {
+	var peerIDStr string
+	multiaddr.ForEach(m, func(c multiaddr.Component) bool {
+		if c.Protocol().Code == multiaddr.P_P2P {
+			peerIDStr = c.Value()
+		}
+		return true
+	})
+	if peerIDStr == "" {
+		panic(multiaddr.ErrProtocolNotFound)
+	}
+	peerID, err := peer.Decode(peerIDStr)
+	if err != nil {
+		panic(err)
+	}
+	return peerID
+}

test/cli/harness/log.go

@@ -0,0 +1,155 @@
+package harness
+
+import (
+	"fmt"
+	"path/filepath"
+	"runtime"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+type event struct {
+	timestamp time.Time
+	msg       string
+}
+
+type events []*event
+
+func (e events) Len() int           { return len(e) }
+func (e events) Less(i, j int) bool { return e[i].timestamp.Before(e[j].timestamp) }
+func (e events) Swap(i, j int)      { e[i], e[j] = e[j], e[i] }
+
+// TestLogger is a logger for tests.
+// It buffers output and only writes the output if the test fails or output is explicitly turned on.
+// The purpose of this logger is to allow Go test to run with the verbose flag without printing logs.
+// The verbose flag is useful since it streams test progress, but also printing logs makes the output too verbose.
+//
+// You can also add prefixes that are prepended to each log message, for extra logging context.
+//
+// This is implemented as a hierarchy of loggers, with children flushing log entries back to parents.
+// This works because t.Cleanup() processes entries in LIFO order, so children always flush first.
+//
+// Obviously this logger should never be used in production systems.
+type TestLogger struct {
+	parent        *TestLogger
+	children      []*TestLogger
+	prefixes      []string
+	prefixesIface []any
+	t             *testing.T
+	buf           events
+	m             sync.Mutex
+	logsEnabled   bool
+}
+
+func NewTestLogger(t *testing.T) *TestLogger {
+	l := &TestLogger{t: t, buf: make(events, 0)}
+	t.Cleanup(l.flush)
+	return l
+}
+
+func (t *TestLogger) buildPrefix(timestamp time.Time) string {
+	d := timestamp.Format("2006-01-02T15:04:05.999999")
+	_, file, lineno, _ := runtime.Caller(2)
+	file = filepath.Base(file)
+	caller := fmt.Sprintf("%s:%d", file, lineno)
+
+	if len(t.prefixes) == 0 {
+		return fmt.Sprintf("%s\t%s\t", d, caller)
+	}
+
+	prefixes := strings.Join(t.prefixes, ":")
+	return fmt.Sprintf("%s\t%s\t%s: ", d, caller, prefixes)
+}
+
+func (t *TestLogger) Log(args ...any) {
+	timestamp := time.Now()
+	e := t.buildPrefix(timestamp) + fmt.Sprint(args...)
+	t.add(&event{timestamp: timestamp, msg: e})
+}
+
+func (t *TestLogger) Logf(format string, args ...any) {
+	timestamp := time.Now()
+	e := t.buildPrefix(timestamp) + fmt.Sprintf(format, args...)
+	t.add(&event{timestamp: timestamp, msg: e})
+}
+
+func (t *TestLogger) Fatal(args ...any) {
+	timestamp := time.Now()
+	e := t.buildPrefix(timestamp) + fmt.Sprint(append([]any{"fatal: "}, args...)...)
+	t.add(&event{timestamp: timestamp, msg: e})
+	t.t.FailNow()
+}
+
+func (t *TestLogger) Fatalf(format string, args ...any) {
+	timestamp := time.Now()
+	e := t.buildPrefix(timestamp) + fmt.Sprintf(fmt.Sprintf("fatal: %s", format), args...)
+	t.add(&event{timestamp: timestamp, msg: e})
+	t.t.FailNow()
+}
+
+func (t *TestLogger) add(e *event) {
+	t.m.Lock()
+	defer t.m.Unlock()
+	t.buf = append(t.buf, e)
+}
+
+func (t *TestLogger) AddPrefix(prefix string) *TestLogger {
+	l := &TestLogger{
+		prefixes:      append(t.prefixes, prefix),
+		prefixesIface: append(t.prefixesIface, prefix),
+		t:             t.t,
+		parent:        t,
+		logsEnabled:   t.logsEnabled,
+	}
+	t.m.Lock()
+	defer t.m.Unlock()
+
+	t.children = append(t.children, l)
+	t.t.Cleanup(l.flush)
+
+	return l
+}
+
+func (t *TestLogger) EnableLogs() {
+	t.m.Lock()
+	defer t.m.Unlock()
+	t.logsEnabled = true
+	if t.parent != nil {
+		if t.parent.logsEnabled {
+			t.parent.EnableLogs()
+		}
+	}
+	fmt.Printf("enabling %d children\n", len(t.children))
+	for _, c := range t.children {
+		if !c.logsEnabled {
+			c.EnableLogs()
+		}
+	}
+}
+
+func (t *TestLogger) flush() {
+	if t.t.Failed() || t.logsEnabled {
+		t.m.Lock()
+		defer t.m.Unlock()
+		// if this is a child, send the events to the parent
+		// the root parent will print all the events in sorted order
+		if t.parent != nil {
+			for _, e := range t.buf {
+				t.parent.add(e)
+			}
+		} else {
+			// we're the root, sort all the events and then print them
+			sort.Sort(t.buf)
+			fmt.Println()
+			fmt.Printf("Logs for test %q:\n\n", t.t.Name())
+			for _, e := range t.buf {
+				fmt.Println(e.msg)
+			}
+			fmt.Println()
+		}
+		t.buf = nil
+	}
+}

test/cli/harness/node.go

@@ -453,9 +453,8 @@ func (n *Node) Peers() []multiaddr.Multiaddr {
 		Path: n.IPFSBin,
 		Args: []string{"swarm", "peers"},
 	})
-	lines := strings.Split(strings.TrimSpace(res.Stdout.String()), "\n")
 	var addrs []multiaddr.Multiaddr
-	for _, line := range lines {
+	for _, line := range res.Stdout.Lines() {
 		ma, err := multiaddr.NewMultiaddr(line)
 		if err != nil {
 			panic(err)
@@ -465,6 +464,28 @@ func (n *Node) Peers() []multiaddr.Multiaddr {
 	return addrs
 }
 
+func (n *Node) PeerWith(other *Node) {
+	n.UpdateConfig(func(cfg *config.Config) {
+		var addrs []multiaddr.Multiaddr
+		for _, addrStr := range other.ReadConfig().Addresses.Swarm {
+			ma, err := multiaddr.NewMultiaddr(addrStr)
+			if err != nil {
+				panic(err)
+			}
+			addrs = append(addrs, ma)
+		}
+
+		cfg.Peering.Peers = append(cfg.Peering.Peers, peer.AddrInfo{
+			ID:    other.PeerID(),
+			Addrs: addrs,
+		})
+	})
+}
+
+func (n *Node) Disconnect(other *Node) {
+	n.IPFS("swarm", "disconnect", "/p2p/"+other.PeerID().String())
+}
+
 // GatewayURL waits for the gateway file and then returns its contents or times out.
 func (n *Node) GatewayURL() string {
 	timer := time.NewTimer(1 * time.Second)

test/cli/harness/nodes.go

@@ -3,6 +3,7 @@ package harness
 import (
 	"sync"
 
+	. "github.com/ipfs/kubo/test/cli/testutils"
 	"github.com/multiformats/go-multiaddr"
 	"golang.org/x/sync/errgroup"
 )
@@ -11,9 +12,7 @@ import (
 type Nodes []*Node
 
 func (n Nodes) Init(args ...string) Nodes {
-	for _, node := range n {
-		node.Init()
-	}
+	ForEachPar(n, func(node *Node) { node.Init(args...) })
 	return n
 }
 
@@ -59,22 +58,11 @@ func (n Nodes) Connect() Nodes {
 }
 
 func (n Nodes) StartDaemons() Nodes {
-	wg := sync.WaitGroup{}
-	for _, node := range n {
-		wg.Add(1)
-		node := node
-		go func() {
-			defer wg.Done()
-			node.StartDaemon()
-		}()
-	}
-	wg.Wait()
+	ForEachPar(n, func(node *Node) { node.StartDaemon() })
 	return n
 }
 
 func (n Nodes) StopDaemons() Nodes {
-	for _, node := range n {
-		node.StopDaemon()
-	}
+	ForEachPar(n, func(node *Node) { node.StopDaemon() })
 	return n
 }