text:
package observer

import (
	"bytes"
	"context"
	"fmt"
	"io"
	"net"
	"testing"

	flowpb "github.com/cilium/cilium/api/v1/flow"
	observerpb "github.com/cilium/cilium/api/v1/observer"
	relaypb "github.com/cilium/cilium/api/v1/relay"
	"github.com/cilium/cilium/pkg/hubble/defaults"
	peerTypes "github.com/cilium/cilium/pkg/hubble/peer/types"
	poolTypes "github.com/cilium/cilium/pkg/hubble/relay/pool/types"
	"github.com/cilium/cilium/pkg/hubble/testutils"

	"github.com/google/go-cmp/cmp"
	"github.com/google/go-cmp/cmp/cmpopts"
	"github.com/sirupsen/logrus"
	"github.com/stretchr/testify/assert"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/connectivity"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/wrapperspb"
)

func TestGetFlows(t *testing.T) {
	type results struct {
		numFlows     int
		flows        map[string][]*flowpb.Flow
		statusEvents []*relaypb.NodeStatusEvent
	}
	var got *results
	type want struct {
		flows        map[string][]*flowpb.Flow
		statusEvents []*relaypb.NodeStatusEvent
		err          error
		log          []string
	}
	fss := &testutils.FakeGRPCServerStream{
		OnContext: context.TODO,
	}
	done := make(chan struct{})
	tests := []struct {
		name   string
		plr    PeerListReporter
		ocb    observerClientBuilder
		req    *observerpb.GetFlowsRequest
		stream observerpb.Observer_GetFlowsServer
		want   want
	}{
		{
			name: "Observe 0 flows from 1 peer without address",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name:    "noip",
								Address: nil,
							},
							Conn: nil,
						},
					}
				},
				OnReportOffline: func(name string) {},
			},
			ocb: fakeObserverClientBuilder{},
			req: &observerpb.GetFlowsRequest{Number: 0},
			stream: &testutils.FakeGetFlowsServer{
				FakeGRPCServerStream: fss,
				OnSend: func(resp *observerpb.GetFlowsResponse) error {
					if resp == nil {
						return nil
					}
					switch resp.GetResponseTypes().(type) {
					case *observerpb.GetFlowsResponse_Flow:
						got.numFlows++
						got.flows[resp.GetNodeName()] = append(got.flows[resp.GetNodeName()], resp.GetFlow())
					case *observerpb.GetFlowsResponse_NodeStatus:
						got.statusEvents = append(got.statusEvents, resp.GetNodeStatus())
					}
					if got.numFlows == 0 && len(got.statusEvents) == 1 {
						close(done)
						return io.EOF
					}
					return nil
				},
			},
			want: want{
				flows: map[string][]*flowpb.Flow{},
				statusEvents: []*relaypb.NodeStatusEvent{
					{
						StateChange: relaypb.NodeState_NODE_UNAVAILABLE,
						NodeNames:   []string{"noip"},
					},
				},
				err: io.EOF,
			},
		}, {
			name: "Observe 4 flows from 2 online peers",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						},
					}
				},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					var numRecv uint64
					return &testutils.FakeObserverClient{
						OnGetFlows: func(_ context.Context, in *observerpb.GetFlowsRequest, _ ...grpc.CallOption) (observerpb.Observer_GetFlowsClient, error) {
							return &testutils.FakeGetFlowsClient{
								OnRecv: func() (*observerpb.GetFlowsResponse, error) {
									if numRecv == in.Number {
										return nil, io.EOF
									}
									numRecv++
									return &observerpb.GetFlowsResponse{
										NodeName: p.Name,
										ResponseTypes: &observerpb.GetFlowsResponse_Flow{
											Flow: &flowpb.Flow{
												NodeName: p.Name,
											},
										},
									}, nil
								},
							}, nil
						},
					}
				},
			},
			req: &observerpb.GetFlowsRequest{Number: 2},
			stream: &testutils.FakeGetFlowsServer{
				FakeGRPCServerStream: fss,
				OnSend: func(resp *observerpb.GetFlowsResponse) error {
					if resp == nil {
						return nil
					}
					switch resp.GetResponseTypes().(type) {
					case *observerpb.GetFlowsResponse_Flow:
						got.numFlows++
						got.flows[resp.GetNodeName()] = append(got.flows[resp.GetNodeName()], resp.GetFlow())
					case *observerpb.GetFlowsResponse_NodeStatus:
						got.statusEvents = append(got.statusEvents, resp.GetNodeStatus())
					}
					if got.numFlows == 4 && len(got.statusEvents) == 1 {
						close(done)
						return io.EOF
					}
					return nil
				},
			},
			want: want{
				flows: map[string][]*flowpb.Flow{
					"one": {&flowpb.Flow{NodeName: "one"}, &flowpb.Flow{NodeName: "one"}},
					"two": {&flowpb.Flow{NodeName: "two"}, &flowpb.Flow{NodeName: "two"}},
				},
				statusEvents: []*relaypb.NodeStatusEvent{
					{
						StateChange: relaypb.NodeState_NODE_CONNECTED,
						NodeNames:   []string{"one", "two"},
					},
				},
				err: io.EOF,
			},
		}, {
			name: "Observe 2 flows from 1 online peer and none from 1 unavailable peer",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.TransientFailure
								},
							},
						},
					}
				},
				OnReportOffline: func(name string) {},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					var numRecv uint64
					return &testutils.FakeObserverClient{
						OnGetFlows: func(_ context.Context, in *observerpb.GetFlowsRequest, _ ...grpc.CallOption) (observerpb.Observer_GetFlowsClient, error) {
							if p.Name != "one" {
								return nil, fmt.Errorf("GetFlows() called for peer '%s'; this is unexpected", p.Name)
							}
							return &testutils.FakeGetFlowsClient{
								OnRecv: func() (*observerpb.GetFlowsResponse, error) {
									if numRecv == in.Number {
										return nil, io.EOF
									}
									numRecv++
									return &observerpb.GetFlowsResponse{
										NodeName: p.Name,
										ResponseTypes: &observerpb.GetFlowsResponse_Flow{
											Flow: &flowpb.Flow{
												NodeName: p.Name,
											},
										},
									}, nil
								},
							}, nil
						},
					}
				},
			},
			req: &observerpb.GetFlowsRequest{Number: 2},
			stream: &testutils.FakeGetFlowsServer{
				FakeGRPCServerStream: fss,
				OnSend: func(resp *observerpb.GetFlowsResponse) error {
					if resp == nil {
						return nil
					}
					switch resp.GetResponseTypes().(type) {
					case *observerpb.GetFlowsResponse_Flow:
						got.numFlows++
						got.flows[resp.GetNodeName()] = append(got.flows[resp.GetNodeName()], resp.GetFlow())
					case *observerpb.GetFlowsResponse_NodeStatus:
						got.statusEvents = append(got.statusEvents, resp.GetNodeStatus())
					}
					if got.numFlows == 2 && len(got.statusEvents) == 2 {
						close(done)
						return io.EOF
					}
					return nil
				},
			},
			want: want{
				flows: map[string][]*flowpb.Flow{
					"one": {&flowpb.Flow{NodeName: "one"}, &flowpb.Flow{NodeName: "one"}},
				},
				statusEvents: []*relaypb.NodeStatusEvent{
					{
						StateChange: relaypb.NodeState_NODE_CONNECTED,
						NodeNames:   []string{"one"},
					}, {
						StateChange: relaypb.NodeState_NODE_UNAVAILABLE,
						NodeNames:   []string{"two"},
					},
				},
				err: io.EOF,
				log: []string{
					`level=info msg="No connection to peer two, skipping" address="192.0.2.2:4244"`,
				},
			},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got = &results{
				flows: make(map[string][]*flowpb.Flow),
			}
			done = make(chan struct{})
			var buf bytes.Buffer
			formatter := &logrus.TextFormatter{
				DisableColors:    true,
				DisableTimestamp: true,
			}
			logger := logrus.New()
			logger.SetOutput(&buf)
			logger.SetFormatter(formatter)
			logger.SetLevel(logrus.DebugLevel)

			srv, err := NewServer(
				tt.plr,
				WithLogger(logger),
				withObserverClientBuilder(tt.ocb),
			)
			assert.NoError(t, err)
			err = srv.GetFlows(tt.req, tt.stream)
			<-done
			assert.Equal(t, tt.want.err, err)
			if diff := cmp.Diff(tt.want.flows, got.flows, cmpopts.IgnoreUnexported(flowpb.Flow{})); diff != "" {
				t.Errorf("Flows mismatch (-want +got):\n%s", diff)
			}
			if diff := cmp.Diff(tt.want.statusEvents, got.statusEvents, cmpopts.IgnoreUnexported(relaypb.NodeStatusEvent{})); diff != "" {
				t.Errorf("StatusEvents mismatch (-want +got):\n%s", diff)
			}
			out := buf.String()
			for _, msg := range tt.want.log {
				assert.Contains(t, out, msg)
			}
		})
	}
}

func TestGetNodes(t *testing.T) {
	type want struct {
		resp *observerpb.GetNodesResponse
		err  error
		log  []string
	}
	tests := []struct {
		name string
		plr  PeerListReporter
		ocb  observerClientBuilder
		req  *observerpb.GetNodesRequest
		want want
	}{
		{
			name: "1 peer without address",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name:    "noip",
								Address: nil,
							},
							Conn: nil,
						},
					}
				},
				OnReportOffline: func(_ string) {},
			},
			ocb: fakeObserverClientBuilder{},
			want: want{
				resp: &observerpb.GetNodesResponse{
					Nodes: []*observerpb.Node{
						{
							Name:    "noip",
							Version: "",
							Address: "",
							State:   relaypb.NodeState_NODE_UNAVAILABLE,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						},
					},
				},
				log: []string{
					`level=info msg="No connection to peer noip, skipping" address="<nil>"`,
				},
			},
		}, {
			name: "2 connected peers",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						},
					}
				},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					return &testutils.FakeObserverClient{
						OnServerStatus: func(_ context.Context, in *observerpb.ServerStatusRequest, _ ...grpc.CallOption) (*observerpb.ServerStatusResponse, error) {
							switch p.Name {
							case "one":
								return &observerpb.ServerStatusResponse{
									UptimeNs:  123456,
									Version:   "cilium v1.9.0",
									MaxFlows:  4095,
									NumFlows:  4095,
									SeenFlows: 11000,
								}, nil
							case "two":
								return &observerpb.ServerStatusResponse{
									UptimeNs:  555555,
									Version:   "cilium v1.9.0",
									MaxFlows:  2047,
									NumFlows:  2020,
									SeenFlows: 12000,
								}, nil
							default:
								return nil, io.EOF
							}
						},
					}
				},
			},
			want: want{
				resp: &observerpb.GetNodesResponse{
					Nodes: []*observerpb.Node{
						{
							Name:      "one",
							Version:   "cilium v1.9.0",
							Address:   "192.0.2.1:4244",
							State:     relaypb.NodeState_NODE_CONNECTED,
							UptimeNs:  123456,
							MaxFlows:  4095,
							NumFlows:  4095,
							SeenFlows: 11000,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						}, {
							Name:      "two",
							Version:   "cilium v1.9.0",
							Address:   "192.0.2.2:4244",
							State:     relaypb.NodeState_NODE_CONNECTED,
							UptimeNs:  555555,
							MaxFlows:  2047,
							NumFlows:  2020,
							SeenFlows: 12000,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						},
					},
				},
			},
		}, {
			name: "2 connected peers with TLS",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
								TLSEnabled:    true,
								TLSServerName: "one.default.hubble-grpc.cilium.io",
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
								TLSEnabled:    true,
								TLSServerName: "two.default.hubble-grpc.cilium.io",
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						},
					}
				},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					return &testutils.FakeObserverClient{
						OnServerStatus: func(_ context.Context, in *observerpb.ServerStatusRequest, _ ...grpc.CallOption) (*observerpb.ServerStatusResponse, error) {
							switch p.Name {
							case "one":
								return &observerpb.ServerStatusResponse{
									UptimeNs:  123456,
									Version:   "cilium v1.9.0",
									MaxFlows:  4095,
									NumFlows:  4095,
									SeenFlows: 11000,
								}, nil
							case "two":
								return &observerpb.ServerStatusResponse{
									UptimeNs:  555555,
									Version:   "cilium v1.9.0",
									MaxFlows:  2047,
									NumFlows:  2020,
									SeenFlows: 12000,
								}, nil
							default:
								return nil, io.EOF
							}
						},
					}
				},
			},
			want: want{
				resp: &observerpb.GetNodesResponse{
					Nodes: []*observerpb.Node{
						{
							Name:      "one",
							Version:   "cilium v1.9.0",
							Address:   "192.0.2.1:4244",
							State:     relaypb.NodeState_NODE_CONNECTED,
							UptimeNs:  123456,
							MaxFlows:  4095,
							NumFlows:  4095,
							SeenFlows: 11000,
							Tls: &observerpb.TLS{
								Enabled:    true,
								ServerName: "one.default.hubble-grpc.cilium.io",
							},
						}, {
							Name:      "two",
							Version:   "cilium v1.9.0",
							Address:   "192.0.2.2:4244",
							State:     relaypb.NodeState_NODE_CONNECTED,
							UptimeNs:  555555,
							MaxFlows:  2047,
							NumFlows:  2020,
							SeenFlows: 12000,
							Tls: &observerpb.TLS{
								Enabled:    true,
								ServerName: "two.default.hubble-grpc.cilium.io",
							},
						},
					},
				},
			},
		}, {
			name: "1 connected peer, 1 unreachable peer",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.TransientFailure
								},
							},
						},
					}
				},
				OnReportOffline: func(_ string) {},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					return &testutils.FakeObserverClient{
						OnServerStatus: func(_ context.Context, in *observerpb.ServerStatusRequest, _ ...grpc.CallOption) (*observerpb.ServerStatusResponse, error) {
							switch p.Name {
							case "one":
								return &observerpb.ServerStatusResponse{
									UptimeNs:  123456,
									Version:   "cilium v1.9.0",
									MaxFlows:  4095,
									NumFlows:  4095,
									SeenFlows: 11000,
								}, nil
							default:
								return nil, io.EOF
							}
						},
					}
				},
			},
			want: want{
				resp: &observerpb.GetNodesResponse{
					Nodes: []*observerpb.Node{
						{
							Name:      "one",
							Version:   "cilium v1.9.0",
							Address:   "192.0.2.1:4244",
							State:     relaypb.NodeState_NODE_CONNECTED,
							UptimeNs:  123456,
							MaxFlows:  4095,
							NumFlows:  4095,
							SeenFlows: 11000,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						}, {
							Name:     "two",
							Version:  "",
							Address:  "192.0.2.2:4244",
							State:    relaypb.NodeState_NODE_UNAVAILABLE,
							UptimeNs: 0,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						},
					},
				},
				log: []string{
					`level=info msg="No connection to peer two, skipping" address="192.0.2.2:4244"`,
				},
			},
		}, {
			name: "1 connected peer, 1 unreachable peer, 1 peer with error",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.TransientFailure
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "three",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.3"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						},
					}
				},
				OnReportOffline: func(_ string) {},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					return &testutils.FakeObserverClient{
						OnServerStatus: func(_ context.Context, in *observerpb.ServerStatusRequest, _ ...grpc.CallOption) (*observerpb.ServerStatusResponse, error) {
							switch p.Name {
							case "one":
								return &observerpb.ServerStatusResponse{
									UptimeNs:  123456,
									Version:   "cilium v1.9.0",
									MaxFlows:  4095,
									NumFlows:  4095,
									SeenFlows: 11000,
								}, nil
							case "three":
								return nil, status.Errorf(codes.Unimplemented, "ServerStatus not implemented")
							default:
								return nil, io.EOF
							}
						},
					}
				},
			},
			want: want{
				resp: &observerpb.GetNodesResponse{
					Nodes: []*observerpb.Node{
						{
							Name:      "one",
							Version:   "cilium v1.9.0",
							Address:   "192.0.2.1:4244",
							UptimeNs:  123456,
							MaxFlows:  4095,
							NumFlows:  4095,
							SeenFlows: 11000,
							State:     relaypb.NodeState_NODE_CONNECTED,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						}, {
							Name:    "two",
							Version: "",
							Address: "192.0.2.2:4244",
							State:   relaypb.NodeState_NODE_UNAVAILABLE,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						}, {
							Name:    "three",
							Version: "",
							Address: "192.0.2.3:4244",
							State:   relaypb.NodeState_NODE_ERROR,
							Tls: &observerpb.TLS{
								Enabled:    false,
								ServerName: "",
							},
						},
					},
				},
				log: []string{
					`level=info msg="No connection to peer two, skipping" address="192.0.2.2:4244"`,
					`level=warning msg="Failed to retrieve server status" error="rpc error: code = Unimplemented desc = ServerStatus not implemented" peer=three`,
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var buf bytes.Buffer
			formatter := &logrus.TextFormatter{
				DisableColors:    true,
				DisableTimestamp: true,
			}
			logger := logrus.New()
			logger.SetOutput(&buf)
			logger.SetFormatter(formatter)
			logger.SetLevel(logrus.DebugLevel)

			srv, err := NewServer(
				tt.plr,
				WithLogger(logger),
				withObserverClientBuilder(tt.ocb),
			)
			assert.NoError(t, err)
			got, err := srv.GetNodes(context.Background(), tt.req)
			assert.Equal(t, tt.want.err, err)
			assert.Equal(t, tt.want.resp, got)
			out := buf.String()
			for _, msg := range tt.want.log {
				assert.Contains(t, out, msg)
			}
		})
	}
}

func TestServerStatus(t *testing.T) {
	type want struct {
		resp *observerpb.ServerStatusResponse
		err  error
		log  []string
	}
	tests := []struct {
		name string
		plr  PeerListReporter
		ocb  observerClientBuilder
		req  *observerpb.ServerStatusRequest
		want want
	}{
		{
			name: "1 peer without address",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name:    "noip",
								Address: nil,
							},
							Conn: nil,
						},
					}
				},
				OnReportOffline: func(_ string) {},
			},
			ocb: fakeObserverClientBuilder{},
			want: want{
				resp: &observerpb.ServerStatusResponse{
					Version:             "hubble-relay",
					NumFlows:            0,
					MaxFlows:            0,
					SeenFlows:           0,
					UptimeNs:            0,
					NumConnectedNodes:   &wrapperspb.UInt32Value{Value: 0},
					NumUnavailableNodes: &wrapperspb.UInt32Value{Value: 1},
					UnavailableNodes:    []string{"noip"},
				},
				log: []string{
					`level=info msg="No connection to peer noip, skipping" address="<nil>"`,
				},
			},
		}, {
			name: "2 connected peers",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						},
					}
				},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					return &testutils.FakeObserverClient{
						OnServerStatus: func(_ context.Context, in *observerpb.ServerStatusRequest, _ ...grpc.CallOption) (*observerpb.ServerStatusResponse, error) {
							switch p.Name {
							case "one":
								return &observerpb.ServerStatusResponse{
									NumFlows:  1111,
									MaxFlows:  1111,
									SeenFlows: 1111,
									UptimeNs:  111111111,
								}, nil
							case "two":
								return &observerpb.ServerStatusResponse{
									NumFlows:  2222,
									MaxFlows:  2222,
									SeenFlows: 2222,
									UptimeNs:  222222222,
								}, nil
							default:
								return nil, io.EOF
							}
						},
					}
				},
			},
			want: want{
				resp: &observerpb.ServerStatusResponse{
					Version:             "hubble-relay",
					NumFlows:            3333,
					MaxFlows:            3333,
					SeenFlows:           3333,
					UptimeNs:            111111111,
					NumConnectedNodes:   &wrapperspb.UInt32Value{Value: 2},
					NumUnavailableNodes: &wrapperspb.UInt32Value{Value: 0},
				},
			},
		}, {
			name: "1 connected peer, 1 unreachable peer",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.Ready
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.TransientFailure
								},
							},
						},
					}
				},
				OnReportOffline: func(_ string) {},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					return &testutils.FakeObserverClient{
						OnServerStatus: func(_ context.Context, in *observerpb.ServerStatusRequest, _ ...grpc.CallOption) (*observerpb.ServerStatusResponse, error) {
							switch p.Name {
							case "one":
								return &observerpb.ServerStatusResponse{
									NumFlows:  1111,
									MaxFlows:  1111,
									SeenFlows: 1111,
									UptimeNs:  111111111,
								}, nil
							default:
								return nil, io.EOF
							}
						},
					}
				},
			},
			want: want{
				resp: &observerpb.ServerStatusResponse{
					Version:             "hubble-relay",
					NumFlows:            1111,
					MaxFlows:            1111,
					SeenFlows:           1111,
					UptimeNs:            111111111,
					NumConnectedNodes:   &wrapperspb.UInt32Value{Value: 1},
					NumUnavailableNodes: &wrapperspb.UInt32Value{Value: 1},
					UnavailableNodes:    []string{"two"},
				},
				log: []string{
					`level=info msg="No connection to peer two, skipping" address="192.0.2.2:4244"`,
				},
			},
		}, {
			name: "2 unreachable peers",
			plr: &testutils.FakePeerListReporter{
				OnList: func() []poolTypes.Peer {
					return []poolTypes.Peer{
						{
							Peer: peerTypes.Peer{
								Name: "one",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.1"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.TransientFailure
								},
							},
						}, {
							Peer: peerTypes.Peer{
								Name: "two",
								Address: &net.TCPAddr{
									IP:   net.ParseIP("192.0.2.2"),
									Port: defaults.ServerPort,
								},
							},
							Conn: &testutils.FakeClientConn{
								OnGetState: func() connectivity.State {
									return connectivity.TransientFailure
								},
							},
						},
					}
				},
				OnReportOffline: func(_ string) {},
			},
			ocb: fakeObserverClientBuilder{
				onObserverClient: func(p *poolTypes.Peer) observerpb.ObserverClient {
					return &testutils.FakeObserverClient{
						OnServerStatus: func(_ context.Context, in *observerpb.ServerStatusRequest, _ ...grpc.CallOption) (*observerpb.ServerStatusResponse, error) {
							return nil, io.EOF
						},
					}
				},
			},
			want: want{
				resp: &observerpb.ServerStatusResponse{
					Version:             "hubble-relay",
					NumFlows:            0,
					MaxFlows:            0,
					SeenFlows:           0,
					UptimeNs:            0,
					NumConnectedNodes:   &wrapperspb.UInt32Value{Value: 0},
					NumUnavailableNodes: &wrapperspb.UInt32Value{Value: 2},
					UnavailableNodes:    []string{"one", "two"},
				},
				log: []string{
					`level=info msg="No connection to peer one, skipping" address="192.0.2.1:4244"`,
					`level=info msg="No connection to peer two, skipping" address="192.0.2.2:4244"`,
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var buf bytes.Buffer
			formatter := &logrus.TextFormatter{
				DisableColors:    true,
				DisableTimestamp: true,
			}
			logger := logrus.New()
			logger.SetOutput(&buf)
			logger.SetFormatter(formatter)
			logger.SetLevel(logrus.DebugLevel)

			srv, err := NewServer(
				tt.plr,
				WithLogger(logger),
				withObserverClientBuilder(tt.ocb),
			)
			assert.NoError(t, err)
			got, err := srv.ServerStatus(context.Background(), tt.req)
			assert.Equal(t, tt.want.err, err)
			assert.Equal(t, tt.want.resp, got)
			out := buf.String()
			for _, msg := range tt.want.log {
				assert.Contains(t, out, msg)
			}
		})
	}
}

type fakeObserverClientBuilder struct {
	onObserverClient func(*poolTypes.Peer) observerpb.ObserverClient
}

func (b fakeObserverClientBuilder) observerClient(p *poolTypes.Peer) observerpb.ObserverClient {
	if b.onObserverClient != nil {
		return b.onObserverClient(p)
	}
	panic("OnObserverClient not set")
}


text:
package datapipelineiface_test

import (
	"testing"

	"github.com/gunosy/aws-sdk-go/service/datapipeline"
	"github.com/gunosy/aws-sdk-go/service/datapipeline/datapipelineiface"
	"github.com/stretchr/testify/assert"
)

func TestInterface(t *testing.T) {
	assert.Implements(t, (*datapipelineiface.DataPipelineAPI)(nil), datapipeline.New(nil))
}

redpajama_set_name: RedPajamaGithub


text:
shop.controller('shopWarehouseControl',['$scope','$http','alertService',function($scope, $http, alertService){

    $scope.csrf = document.getElementById('csrf').value;
    $scope.alertClass = alertService.alertBox;
    $scope.alertText = alertService.alertBox;
    $scope.formTitle = warehouse.formTitle + " " + headers.warehouse;
    $scope.loading = false;
    $scope.name = null;
    $scope.noField = null;
    $scope.orderByField = null;
    $scope.request = null;
    $scope.resultTitle = search.resultTitle;
    $scope.reverseSort  = false
    $scope.searchButton = search.searchButton;
    $scope.searchClear = search.clear;
    $scope.searchOption = null;
    $scope.searchRadio = null;
    $scope.searchResponse = null;
    $scope.searchSave = search.save;
    $scope.searchText = null;
    $scope.searchTitle = search.searchTitle + " " + headers.warehouse;
    $scope.searchType = [];
    $scope.searchType.push({option: 'name', name: warehouseSearch.nameSearchTitle, text: ''});
    $scope.showForm = false;
    $scope.showQuery = false;
    $scope.showSearch = false;
    $scope.warehouseId = 0;
    $scope.warehouseName = warehouse.name;
    $scope.tooltip = {
        'form': actions.extend,
        'search': actions.extend,
        'query': actions.extend
    };
    $scope.url = null;

    $scope.clearWarehouse = function(){
        $scope.warehouseId = 0;
        $scope.name = null;
    };

    $scope.clearQuery = function(){
        $scope.searchResponse = null;
        $scope.searchType = [];
        $scope.searchType.push({option: 'name', name: warehouseSearch.nameSearchTitle, text: ''});
    };

    $scope.getTitle = function(formStatus){
        if (formStatus == true){
            return actions.collapse;
        }else{
            return actions.extend;
        }
    };

    $scope.insertWarehouse = function(){
        $scope.noField = null;
        $scope.alertText.alert_ = '';
        var new_warehouse = {
            'warehouse_id': $scope.warehouseId,
            'name': $scope.name
        };
        for (item in new_warehouse){
            if (new_warehouse[item] == null || new_warehouse[item] == ''){
                if (item != 'warehouse_id'){
                    $scope.alertText.alert_ = alerts.emptyFields;
                    $scope.alertClass.class_ = 'fail';
                    $scope.noField = item;
                    return null;
                }
            }
        };
        $scope.url = config.apiURL + "/warehouse/insert";
        $http({
                 method: 'POST',
                 url: $scope.url,
                 data: new_warehouse,
                 headers: {'ContentType': 'application/json', 'X-CSRFToken': $scope.csrf}
        }).then(
            function(data){
                    $scope.clearQuery();
                    $scope.alertText.alert_ = alerts.insertWarehouse;
                    $scope.alertClass.class_ = 'win';
                    $scope.clearWarehouse();
            },
            function(){
                $scope.alertText.alert_ = alerts.failedInsert;
                $scope.alertClass.class_ = 'fail';
        });
    };

    $scope.loadWarehouseForm = function(){
        $scope.showSearch = false;
        $scope.showForm = true;
        $scope.warehouseId = this.item.warehouse_id;
        $scope.name = this.item.name;
    };

    $scope.printSearch = function(){
        $scope.showQuery = true;
        $scope.searchResponse = [];
        $scope.alertText.alert_ = '';
        $scope.loading = true;
        if ($scope.searchRadio == null){
            $scope.alertText.alert_ = alerts.noSearch;
            $scope.alertClass.class_ = "fail";
            return false;
        }
        if ($scope.searchText == null || $scope.searchText == ""){
            $scope.searchType.forEach(function(type){
                if (type.name == $scope.searchRadio){
                    $scope.searchText = type.text;
                }
            });
        }
        $scope.request = {
            option: $scope.searchOption,
            warehouse: $scope.searchText
        };
        $scope.url = config.apiURL + "/warehouse/find";
        $http({
                 method: 'POST',
                 url: $scope.url,
                 data: $scope.request,
                 headers: {'ContentType': 'application/json', 'X-CSRFToken': $scope.csrf}
        }).success(
            function(data){
                response = data.warehouses;
                $scope.searchResponse = [];
                for (item in response){
                    $scope.searchResponse.push(response[item]);
                }
                $scope.loading = false;
        }).error(
            function(){
                $scope.alertText.alert_ = alerts.noResults;
                $scope.alertClass.class_ = 'fail';
                $scope.loading = false;
        });
    };

    $scope.setSearch = function(type){
        $scope.searchText = type.text.replace(/\*/g, '%');
        $scope.searchOption = type.option;
        $scope.searchRadio = type.name;
    };

    $scope.toggleContainer = function(container){
        switch (container){
            case 'search':
                $scope.showSearch = !$scope.showSearch;
                $scope.tooltip.search = $scope.getTitle($scope.showSearch);
                $scope.showForm = false;
                $scope.tooltip.form = $scope.getTitle($scope.showForm);
                break;
            case 'form':
                $scope.showForm = !$scope.showForm;
                $scope.tooltip.form = $scope.getTitle($scope.showForm);
                $scope.showSearch = false;
                $scope.tooltip.search = $scope.getTitle($scope.showSearch);
                break;
            case 'query':
                $scope.showQuery = !$scope.showQuery;
                $scope.tooltip.query = $scope.getTitle($scope.showQuery);
                break;
        }
    };
}]);


text:
console.log(state);
// output: undefined

var state = "ready";

/*

等于

 */

var state;
console.log(state);
// output: undefined

state = "ready";


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

<META http-equiv="Content-Type" content="text/html; charset=UTF-8">
<link rel="stylesheet" href="../../../style.css" type="text/css" media="screen">
<link rel="stylesheet" href="../../../print.css" type="text/css" media="print">
<meta content="io,com.yui.util.io,send" name="keywords">
<title>com.yui.util.io</title>

YUI AS Documentation	All Packages | All Classes | Index | FramesNo Frames

Class io	Methods

Package	com.yui.util

Class	public class io

Inheritance	io  flash.display.Sprite

Public Methods

 	Method	Defined by

io()

	io

send(uri:String, cfg:Object, id:Number):void

	io

Constructor detail

io	()	constructor

public function io()

Method detail

send	()	method

public function send(uri:String, cfg:Object, id:Number):voidParameters

	uri:String

	cfg:Object

	id:Number


text:
"""
session.py

Created by Thomas Mangin on 2014-06-22.
Copyright (c) 2014-2015 Exa Networks. All rights reserved.
"""

from exabgp.configuration.engine.raised import Raised
from exabgp.configuration.engine.section import Section
from exabgp.configuration.engine.parser import ip
from exabgp.configuration.engine.parser import holdtime

from exabgp.configuration.bgp.capability import syntax_capability
from exabgp.configuration.bgp.capability import SectionCapability

from exabgp.configuration.bgp.asn import SectionASN
from exabgp.configuration.bgp.asn import syntax_asn

# ============================================================== syntax_session

syntax_session = """\
session <name> {
	%s
	%s
}
""" % (
	'\n\t'.join((_.replace(' <name>','') for _ in syntax_asn.split('\n'))),
	'\n\t'.join((_.replace(' <name>','') for _ in syntax_capability.split('\n'))),
)

# =============================================================== RaisedSession

class RaisedSession (Raised):
	syntax = syntax_session

# ============================================================== SectionSession
#

class SectionSession (Section):
	syntax = syntax_session
	name = 'session'

	def exit (self, tokeniser):
		asn = self.extract_anonymous('asn',tokeniser)
		if asn:
			if 'asn' in self.content:
				raise RaisedSession(tokeniser,"can not have amed and anonymous 'asn' in a session")
			self.content['asn'] = asn

		if 'asn' not in self.content:
			raise RaisedSession(tokeniser,"section is missing a 'asn' section")

		capability = self.extract_anonymous('capability',tokeniser)
		if capability:
			if 'capability' in self.content:
				raise RaisedSession(tokeniser,"can not have amed and anonymous 'capability' in a session")
			self.content['capability'] = capability

		if 'capability' not in self.content:
			raise RaisedSession(tokeniser,"section is missing a 'capability' section")

		if 'router-id' not in self.content:
			# 0.0.0.0 is now a invlid router-id so it will be replaced by the bind ip
			self.content['router-id'] = ip(lambda:'0.0.0.0')

		if 'hold-time' not in self.content:
			self.content['hold-time'] = holdtime(lambda:'180')

	def router_id (self, tokeniser):
		try:
			self.content['router-id'] = ip(tokeniser)
		except ValueError,exc:
			raise RaisedSession(tokeniser,'could not parse router-id, %s' % str(exc))

	def hold_time (self, tokeniser):
		try:
			self.content['hold-time'] = holdtime(tokeniser)
		except ValueError,exc:
			raise RaisedSession(tokeniser,'could not parse hold-time, %s' % str(exc))

	def asn (self, tokeniser):
		section = self.get_section(SectionASN.name,tokeniser)
		if section:
			self.content['asn'] = section
		else:
			return False

	def capability (self, tokeniser):
		section = self.get_section(SectionCapability.name,tokeniser)
		if section:
			self.content['capability'] = section
		else:
			return False

	@classmethod
	def register (cls, registry, location):
		registry.register_class(cls)

		registry.register(SectionASN,location+['asn'])
		registry.register_hook(cls,'action',location+['asn'],'asn')

		registry.register(SectionCapability,location+['capability'])
		registry.register_hook(cls,'action',location+['capability'],'capability')

		registry.register_hook(cls,'enter',location,'enter')
		registry.register_hook(cls,'exit',location,'exit')

		registry.register_hook(cls,'action',location+['router-id'],'router_id')
		registry.register_hook(cls,'action',location+['hold-time'],'hold_time')

		# registry.register_hook(cls,'enter',location,'enter_nameless')
		# registry.register_hook(cls,'action',location+['local'],'local_asn')
		# registry.register_hook(cls,'action',location+['peer'],'peer_asn')
		# registry.register_hook(cls,'exit', location,'exit_nameless')


text:
'use strict';

module.exports = function (listener) {

    // button mouseover and mouseaway selfs
    let canvas = this.network.canvas.frame.canvas;
    let canvasOffset = canvas.getBoundingClientRect();

    canvas.addEventListener('mousemove', event => {

        if (!this.menu) {
            return;
        }

        let mouse = this.network.DOMtoCanvas({
            x: event.offsetX - canvasOffset.left,
            y: event.offsetY - canvasOffset.top
        });

        this.menu.buttons.forEach(button => {

            if (!button.ref) {
                return
            }

            let distance = Math.sqrt((mouse.x - button.ref.x)*(mouse.x - button.ref.x) + (mouse.y - button.ref.y)*(mouse.y - button.ref.y));
            if (distance < button.ref.radius) {
                if (!button.ref.isHover) {
                    button.ref.hover();
                }
            } else {
                if (button.ref.isHover) {
                    button.ref.blur();
                }
            }
        });
    }, false);

    // listen for click events on the buttons
    this.network.on('click', event => {

        if (!this.menu) {
            return;
        }

        let mouse = {
            x: event.pointer.canvas.x,
            y: event.pointer.canvas.y
        };

        this.menu.buttons.forEach(button => {

            if (!button.ref) {
                return
            }

            let distance = Math.sqrt((mouse.x - button.ref.x)*(mouse.x - button.ref.x) + (mouse.y - button.ref.y)*(mouse.y - button.ref.y));
            if (distance < button.ref.radius) {
                listener({
                    name: button.event,
                    event: {
                        node: button.node,
                        edge: button.edge,
                        data: button.data
                    }
                })
            }
        });
    });
};


text:
from freebsd import *
from freebsd.const import *


text:
/**
 * generated by Xtext 2.17.1
 */
package com.specmate.cause_effect_patterns.internal.specDSL;

import org.eclipse.emf.ecore.EObject;

/**
 *
 * A representation of the model object 'Subtree'.
 *
 *
 *

 * The following features are supported:
 *

 *

 *
{@link com.specmate.cause_effect_patterns.specDSL.Subtree#getName Name}

 *

 *
 * @see com.specmate.cause_effect_patterns.specDSL.SpecDSLPackage#getSubtree()
 * @model
 * @generated
 */
public interface Subtree extends EObject
{
  /**
   * Returns the value of the 'Name' reference.
   *
   *

   * If the meaning of the 'Name' reference isn't clear,
   * there really should be more of a description here...
   *

   *
   * @return the value of the 'Name' reference.
   * @see #setName(TreeTag)
   * @see com.specmate.cause_effect_patterns.specDSL.SpecDSLPackage#getSubtree_Name()
   * @model
   * @generated
   */
  TreeTag getName();

  /**
   * Sets the value of the '{@link com.specmate.cause_effect_patterns.specDSL.Subtree#getName Name}' reference.
   *
   *
   * @param value the new value of the 'Name' reference.
   * @see #getName()
   * @generated
   */
  void setName(TreeTag value);

} // Subtree


text:
package io.sigpipe.sing.dataset.feature;

import java.io.IOException;

import java.nio.ByteBuffer;

import io.sigpipe.sing.serialization.ByteSerializable;
import io.sigpipe.sing.serialization.SerializationInputStream;
import io.sigpipe.sing.serialization.SerializationOutputStream;

/**
 * Stores binary feature information.
 *
 * @author malensek
 */
public class BinaryFeatureData
extends FeatureData<ByteArray> implements ByteSerializable {

    public BinaryFeatureData(ByteArray data) {
        super(data);
    }

    public BinaryFeatureData(byte[] data) {
        super(new ByteArray(data));
    }

    @Override
    public int toInt() {
        ByteBuffer buffer = ByteBuffer.wrap(this.data.getBytes());
        return buffer.getInt();
    }

    @Override
    public long toLong() {
        ByteBuffer buffer = ByteBuffer.wrap(this.data.getBytes());
        return buffer.getLong();
    }

    @Override
    public float toFloat() {
        ByteBuffer buffer = ByteBuffer.wrap(this.data.getBytes());
        return buffer.getFloat();
    }

    @Override
    public double toDouble() {
        ByteBuffer buffer = ByteBuffer.wrap(this.data.getBytes());
        return buffer.getDouble();
    }

    @Override
    public String toString() {
        return new String(this.data.getBytes());
    }

    @Override
    public byte[] toBytes() {
        return this.data.getBytes();
    }

    @Deserialize
    public BinaryFeatureData(SerializationInputStream in)
    throws IOException {
        super(new ByteArray(in.readField()));
    }

    @Override
    public void serialize(SerializationOutputStream out)
    throws IOException {
        out.writeField(data.getBytes());
    }
}


text:
package org.eclipse.php.ui.util;

import org.eclipse.jface.viewers.IElementComparer;

/**
 *
 * @author guy.g
 *
 */
public interface IPHPTreeElementComparer extends IElementComparer {

	public boolean supports(Object o);
}