<commit_before>package mocknet

import (
	"testing"
	"time"

	ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
	inet "github.com/ipfs/go-ipfs/p2p/net"
)

func TestNotifications(t *testing.T) {

	mn, err := FullMeshLinked(context.Background(), 5)
	if err != nil {
		t.Fatal(err)
	}

	timeout := 10 * time.Second

	// signup notifs
	nets := mn.Nets()
	notifiees := make([]*netNotifiee, len(nets))
	for i, pn := range nets {
		n := newNetNotifiee()
		pn.Notify(n)
		notifiees[i] = n
	}

	// connect all
	for _, n1 := range nets {
		for _, n2 := range nets {
			if n1 == n2 {
				continue
			}
			if _, err := mn.ConnectNets(n1, n2); err != nil {
				t.Fatal(err)
			}
		}
	}

	// test everyone got the correct connection opened calls
	for i, s := range nets {
		n := notifiees[i]
		for _, s2 := range nets {
			cos := s.ConnsToPeer(s2.LocalPeer())
			func() {
				for i := 0; i < len(cos); i++ {
					var c inet.Conn
					select {
					case c = <-n.connected:
					case <-time.After(timeout):
						t.Fatal("timeout")
					}
					for _, c2 := range cos {
						if c == c2 {
							t.Log("got notif for conn")
							return
						}
					}
					t.Error("connection not found")
				}
			}()
		}
	}

	complement := func(c inet.Conn) (inet.Network, *netNotifiee, *conn) {
		for i, s := range nets {
			for _, c2 := range s.Conns() {
				if c2.(*conn).rconn == c {
					return s, notifiees[i], c2.(*conn)
				}
			}
		}
		t.Fatal("complementary conn not found", c)
		return nil, nil, nil
	}

	testOCStream := func(n *netNotifiee, s inet.Stream) {
		var s2 inet.Stream
		select {
		case s2 = <-n.openedStream:
			t.Log("got notif for opened stream")
		case <-time.After(timeout):
			t.Fatal("timeout")
		}
		if s != nil && s != s2 {
			t.Fatalf("got incorrect stream %p %p", s, s2)
		}

		select {
		case s2 = <-n.closedStream:
			t.Log("got notif for closed stream")
		case <-time.After(timeout):
			t.Fatal("timeout")
		}
		if s != nil && s != s2 {
			t.Fatalf("got incorrect stream %p %p", s, s2)
		}
	}

	streams := make(chan inet.Stream)
	for _, s := range nets {
		s.SetStreamHandler(func(s inet.Stream) {
			streams <- s
			s.Close()
		})
	}

	// there's one stream per conn that we need to drain....
	// unsure where these are coming from
	for i, _ := range nets {
		n := notifiees[i]
		testOCStream(n, nil)<commit_msg>Run 'gofmt -s -w' on these files<commit_after>package mocknet

import (
	"testing"
	"time"

	ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
	inet "github.com/ipfs/go-ipfs/p2p/net"
)

func TestNotifications(t *testing.T) {

	mn, err := FullMeshLinked(context.Background(), 5)
	if err != nil {
		t.Fatal(err)
	}

	timeout := 10 * time.Second

	// signup notifs
	nets := mn.Nets()
	notifiees := make([]*netNotifiee, len(nets))
	for i, pn := range nets {
		n := newNetNotifiee()
		pn.Notify(n)
		notifiees[i] = n
	}

	// connect all
	for _, n1 := range nets {
		for _, n2 := range nets {
			if n1 == n2 {
				continue
			}
			if _, err := mn.ConnectNets(n1, n2); err != nil {
				t.Fatal(err)
			}
		}
	}

	// test everyone got the correct connection opened calls
	for i, s := range nets {
		n := notifiees[i]
		for _, s2 := range nets {
			cos := s.ConnsToPeer(s2.LocalPeer())
			func() {
				for i := 0; i < len(cos); i++ {
					var c inet.Conn
					select {
					case c = <-n.connected:
					case <-time.After(timeout):
						t.Fatal("timeout")
					}
					for _, c2 := range cos {
						if c == c2 {
							t.Log("got notif for conn")
							return
						}
					}
					t.Error("connection not found")
				}
			}()
		}
	}

	complement := func(c inet.Conn) (inet.Network, *netNotifiee, *conn) {
		for i, s := range nets {
			for _, c2 := range s.Conns() {
				if c2.(*conn).rconn == c {
					return s, notifiees[i], c2.(*conn)
				}
			}
		}
		t.Fatal("complementary conn not found", c)
		return nil, nil, nil
	}

	testOCStream := func(n *netNotifiee, s inet.Stream) {
		var s2 inet.Stream
		select {
		case s2 = <-n.openedStream:
			t.Log("got notif for opened stream")
		case <-time.After(timeout):
			t.Fatal("timeout")
		}
		if s != nil && s != s2 {
			t.Fatalf("got incorrect stream %p %p", s, s2)
		}

		select {
		case s2 = <-n.closedStream:
			t.Log("got notif for closed stream")
		case <-time.After(timeout):
			t.Fatal("timeout")
		}
		if s != nil && s != s2 {
			t.Fatalf("got incorrect stream %p %p", s, s2)
		}
	}

	streams := make(chan inet.Stream)
	for _, s := range nets {
		s.SetStreamHandler(func(s inet.Stream) {
			streams <- s
			s.Close()
		})
	}

	// there's one stream per conn that we need to drain....
	// unsure where these are coming from
	for i := range nets {
		n := notifiees[i]
		testOCStream(n, nil)
{'fraction_non_alphanumeric': 0.16331967213114754, 'fraction_numerical': 0.010245901639344262, 'mean_word_length': 2.5216450216450217, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2822711', 'n_tokens_mistral': 2307, 'n_tokens_neox': 1929, 'n_words': 483}