package main import ( "errors" "fmt" "testing" "time" ) 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 TestScenarioUsersScaleWithConfiguredCapacity(t *testing.T) { tests := []struct { users int group int mixedP2P int mixedGroup int }{ {users: 1000, group: 500, mixedP2P: 300, mixedGroup: 250}, {users: 600, group: 300, mixedP2P: 180, mixedGroup: 150}, } for _, tc := range tests { if got := scaledUsers(tc.users, 1, 2); got != tc.group { t.Fatalf("users=%d group=%d want=%d", tc.users, got, tc.group) } if got := scaledUsers(tc.users, 3, 10); got != tc.mixedP2P { t.Fatalf("users=%d mixed p2p=%d want=%d", tc.users, got, tc.mixedP2P) } if got := scaledUsers(tc.users, 1, 4); got != tc.mixedGroup { t.Fatalf("users=%d mixed group=%d want=%d", tc.users, got, tc.mixedGroup) } } } func TestNewHistoryQueryIncludesInterleavedSequenceRange(t *testing.T) { messages := make([]AcceptedMessage, 10) for i := range messages { messages[i] = AcceptedMessage{ClientMid: fmt.Sprintf("mid-%d", i), Seq: int64(100 + i*2)} } query := newHistoryQuery(nil, "p2p-one", messages, 10) if query.since != 100 || query.before != 119 || query.limit != 19 { t.Fatalf("history query range=%d-%d limit=%d want=100-119/19", query.since, query.before, query.limit) } } 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) } } func TestCloseScheduleInterleavesLogicalAgents(t *testing.T) { cfg := &Config{CloseRamp: "60s"} intervalA, delayA := closeSchedule(cfg, "a", 601) intervalB, delayB := closeSchedule(cfg, "b", 601) if intervalA != 200*time.Millisecond || intervalB != intervalA { t.Fatalf("intervals: a=%s b=%s, want 200ms", intervalA, intervalB) } if delayA != 0 || delayB != 100*time.Millisecond { t.Fatalf("delays: a=%s b=%s, want 0/100ms", delayA, delayB) } }