package main import ( "errors" "testing" ) func TestHistoryQueriesForMessagesGroupsRangesByTopic(t *testing.T) { p2pClient := &WSClient{} groupClient := &WSClient{} run := &agentRun{cfg: &Config{HistoryPageSize: 200}, clients: map[int]*WSClient{1: groupClient, 7: p2pClient}} messages := make([]AcceptedMessage, 0, 207) for seq := int64(205); seq >= 1; seq-- { messages = append(messages, AcceptedMessage{Kind: "p2p", Topic: "p2p-a", PeerIndex: 7, Seq: seq}) } messages = append(messages, AcceptedMessage{Kind: "group", Topic: "grp-a", PeerIndex: 99, Seq: 11}, AcceptedMessage{Kind: "group", Topic: "grp-a", PeerIndex: 99, Seq: 12}, ) queries := run.historyQueriesForMessages(messages) if len(queries) != 3 { t.Fatalf("queries=%d want=3", len(queries)) } p2pQueries := 0 groupQueries := 0 for _, query := range queries { switch query.topic { case "p2p-a": p2pQueries++ if query.client != p2pClient || query.limit != 200 || len(query.expected) == 0 { t.Fatalf("unexpected p2p query: %#v", query) } case "grp-a": groupQueries++ if query.client != groupClient || query.since != 11 || query.before != 13 { t.Fatalf("unexpected group query: %#v", query) } } } if p2pQueries != 2 || groupQueries != 1 { t.Fatalf("p2p queries=%d group queries=%d", p2pQueries, groupQueries) } } func TestClassifyPublishErrorTreatsClosedConnectionAsUnknownSocketError(t *testing.T) { run := &agentRun{} run.classifyPublishError(errors.New("write tcp: use of closed network connection")) if run.publishSocket.Load() != 1 || run.publishOther.Load() != 0 { t.Fatalf("socket=%d other=%d", run.publishSocket.Load(), run.publishOther.Load()) } } func TestMergeMessageLedgersPrefersPeerObservationAndAck(t *testing.T) { senderEcho := AcceptedMessage{ClientMid: "mid-1", SenderMachine: "a", PeerMachine: "a", Seq: 10} peerDelivery := AcceptedMessage{ClientMid: "mid-1", SenderMachine: "a", PeerMachine: "b", PeerIndex: 7, Seq: 10} merged := mergeMessageLedgers(nil, []AcceptedMessage{senderEcho, peerDelivery}) if len(merged) != 1 || merged[0].PeerMachine != "b" || merged[0].PeerIndex != 7 { t.Fatalf("peer observation was not preferred: %+v", merged) } ack := AcceptedMessage{ClientMid: "mid-1", SenderMachine: "a", PeerMachine: "b", PeerIndex: 7, Seq: 10, Scenario: "p2p"} merged = mergeMessageLedgers([]AcceptedMessage{ack}, []AcceptedMessage{peerDelivery}) if len(merged) != 1 || merged[0].Scenario != "p2p" { t.Fatalf("ACK ledger was not authoritative: %+v", merged) } }