package distworker import ( "encoding/json" "net/http" "net/http/httptest" "testing" "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "rocketgit.ru/rsmon/worker/internal/wire" ) func TestRunMainAPIHTTPCheck_OK(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() ok := runMainAPIHTTPCheck(srv.URL) assert.True(t, ok, "200 OK should be a successful probe") } func TestRunMainAPIHTTPCheck_404IsDown(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNotFound) })) defer srv.Close() ok := runMainAPIHTTPCheck(srv.URL) assert.False(t, ok, "normal HTTP check flow treats non-200 as down") } func TestRunMainAPIHTTPCheck_5xxIsDown(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) })) defer srv.Close() ok := runMainAPIHTTPCheck(srv.URL) assert.False(t, ok, "5xx should be treated as down") } func TestRunMainAPIHTTPCheck_NetworkErrorIsDown(t *testing.T) { ok := runMainAPIHTTPCheck("http://127.0.0.1:1") assert.False(t, ok, "connection refused should be treated as down") } func TestSelfcheckTarget(t *testing.T) { assert.Equal(t, "https://rsmon.ru/up", selfcheckTarget("https://rsmon.ru")) assert.Equal(t, "https://rsmon.ru/up", selfcheckTarget("https://rsmon.ru/")) assert.Equal(t, "https://api.example.com/api/v1/up", selfcheckTarget("https://api.example.com/api/v1")) assert.Equal(t, "", selfcheckTarget(""), "empty base URL yields empty target") } func TestParseTelegramChatID(t *testing.T) { id, err := parseTelegramChatID("200318758") require.NoError(t, err) assert.Equal(t, int64(200318758), id) _, err = parseTelegramChatID("") assert.Error(t, err) _, err = parseTelegramChatID("not-a-number") assert.Error(t, err) } func TestSMTPCredToModelCarriesPlaintextSecret(t *testing.T) { c := wire.SMTPCredential{ ID: 11, Name: "primary", Server: "smtp.example.com", Port: 587, Login: "alerts@example.com", Password: "smtp-password-xyz", FromName: "RSMon", FromAddress: "alerts@example.com", } cred := smtpCredToModel(&c) require.NotNil(t, cred) assert.Equal(t, "plain:smtp-password-xyz", cred.SecretEnc) assert.Equal(t, int64(11), cred.ID) assert.Equal(t, models_credentialKindSMTP(), cred.Kind) got, err := cred.GetSecret() require.NoError(t, err) assert.Equal(t, "smtp-password-xyz", got, "worker-side credential must yield plaintext via models.GetSecret") } func TestTelegramCredToModelCarriesPlaintextToken(t *testing.T) { c := wire.TelegramCredential{ ID: 22, Name: "main-bot", BotName: "rsmon_alerts_bot", Token: "bot-token-9876543210:ABCDEFG", APIURL: "https://api.telegram.org", } cred := telegramCredToModel(c) require.NotNil(t, cred) assert.Equal(t, "plain:bot-token-9876543210:ABCDEFG", cred.SecretEnc) assert.Equal(t, int64(22), cred.ID) assert.Equal(t, models_credentialKindTelegram(), cred.Kind) got, err := cred.GetSecret() require.NoError(t, err) assert.Equal(t, "bot-token-9876543210:ABCDEFG", got) } // models_credentialKindSMTP/Telegram are local shims to avoid an import-cycle // in this isolated test file. The real constants live in app/models and are // asserted at runtime through the wire translation functions. func models_credentialKindSMTP() string { return "smtp" } func models_credentialKindTelegram() string { return "telegram" } func TestApplyInitStoresSystemContacts(t *testing.T) { r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) assert.Empty(t, r.SystemContacts(), "no system contacts before init") want := []wire.SystemContact{ {ID: 1, Kind: "email", Value: "ops@example.com", Name: "ops"}, {ID: 2, Kind: "telegram_group", Value: "200318758", Name: "alerts"}, } r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Concurrency: 2, SystemContacts: want, }) got := r.SystemContacts() require.Len(t, got, 2) assert.Equal(t, want[0].ID, got[0].ID) assert.Equal(t, "email", got[0].Kind) assert.Equal(t, "ops@example.com", got[0].Value) assert.Equal(t, "telegram_group", got[1].Kind) assert.Equal(t, "200318758", got[1].Value) r.applyInit(&wire.WorkerInit{WorkerID: "w-1", Concurrency: 2}) assert.Empty(t, r.SystemContacts(), "subsequent init with empty SystemContacts should replace cache") } func TestRunSelfcheckOnce_FailureStartsDownStateNoAlert(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) })) defer srv.Close() r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.config.URL = srv.URL creds := &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp.example.com", Port: 587, Login: "a@a", FromAddress: "a@a", Password: "x"}}, } r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: creds, SystemContacts: []wire.SystemContact{{ID: 1, Kind: "email", Value: "ops@example.com"}}, }) state := &selfcheckState{consensus: &consensusState{}} r.runSelfcheckOnce(t.Context(), state) require.NotNil(t, state.consensus.downSince, "first down probe must stamp downSince") assert.False(t, state.consensus.alertActive, "no alert expected before selfcheckConsensusWait elapses") } func TestRunSelfcheckOnce_AlertFiredAfterConsensusWait(t *testing.T) { r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.config.URL = "http://127.0.0.1:1" r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, }) state := &selfcheckState{consensus: &consensusState{}} downSince := time.Now().Add(-selfcheckConsensusWait - time.Second) state.consensus.downSince = &downSince r.runSelfcheckOnce(t.Context(), state) assert.True(t, state.consensus.alertActive, "alert must be active after selfcheckConsensusWait elapses") } func TestRunSelfcheckOnce_RecoveryClearsAlert(t *testing.T) { srvOK := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) defer srvOK.Close() r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, }) state := &selfcheckState{consensus: &consensusState{}} state.consensus.alertActive = true downSince := time.Now().Add(-20 * time.Minute) state.consensus.downSince = &downSince r.config.URL = srvOK.URL // First up tick: marks the up consensus start; recovery alert // waits for selfcheckConsensusWait. Assert the alert is still // active so we can distinguish "up just arrived" from "up held". r.runSelfcheckOnce(t.Context(), state) assert.True(t, state.consensus.alertActive, "recovery alert should not fire on the first up tick") assert.Equal(t, consensusUp, state.consensus.lastVerdict) // Tick again with the up verdict now held past // selfcheckConsensusWait so the recovery alert fires. state.consensus.downSince = ptrTime(time.Now().Add(-selfcheckConsensusWait - time.Second)) r.runSelfcheckOnce(t.Context(), state) assert.False(t, state.consensus.alertActive, "recovery alert must clear alertActive after selfcheckConsensusWait up") assert.Nil(t, state.consensus.downSince) } func TestRunSelfcheckOnce_RequiresSelfVote(t *testing.T) { // When self has not yet produced a verdict (nil), the // consensus must refuse to call the cluster down regardless of // what peers report. up := true peers := []peerObservation{{WorkerID: "p1", Up: true, ObservedAt: time.Now()}} verdict := decideConsensus(nil, peers, 2) assert.Equal(t, consensusNoQuorum, verdict, "missing self vote must prevent a down consensus") verdict = decideConsensus(&up, peers, 2) assert.Equal(t, consensusUp, verdict, "self up + peer up must reach consensus up") } func TestSendSystemAlert_NoCredentials(t *testing.T) { r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", SystemContacts: []wire.SystemContact{{ID: 1, Kind: "email", Value: "ops@example.com"}}, }) r.sendSystemAlert(t.Context(), true, "test message") } func TestSendSystemAlert_NoContacts(t *testing.T) { r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, }) r.sendSystemAlert(t.Context(), true, "test message") } func TestWorkerInit_SystemContactsJSONRoundtrip(t *testing.T) { init := wire.WorkerInit{ WorkerID: "w-1", Concurrency: 4, SystemContacts: []wire.SystemContact{ {ID: 1, Kind: "email", Value: "ops@example.com", Name: "ops"}, {ID: 2, Kind: "telegram_private", Value: "200318758", Name: "alerts"}, }, } raw, err := json.Marshal(init) require.NoError(t, err) assert.Contains(t, string(raw), `"system_contacts":`) var decoded wire.WorkerInit require.NoError(t, json.Unmarshal(raw, &decoded)) require.Len(t, decoded.SystemContacts, 2) assert.Equal(t, init.SystemContacts[0], decoded.SystemContacts[0]) assert.Equal(t, init.SystemContacts[1], decoded.SystemContacts[1]) } func TestWorkerInit_PeersJSONRoundtrip(t *testing.T) { init := wire.WorkerInit{ WorkerID: "w-1", Concurrency: 4, Peers: []wire.PeerInfo{ {WorkerID: "w-2", URL: "http://127.0.0.1:27402", RegionCode: "local"}, {WorkerID: "w-3", URL: "http://127.0.0.1:27403", RegionCode: "local", Login: "ops", Password: "x"}, }, } raw, err := json.Marshal(init) require.NoError(t, err) body := string(raw) assert.Contains(t, body, `"peers":[`) assert.Contains(t, body, `"worker_id":"w-2"`) assert.Contains(t, body, `"url":"http://127.0.0.1:27402"`) assert.Contains(t, body, `"login":"ops"`) assert.Contains(t, body, `"password":"x"`) var decoded wire.WorkerInit require.NoError(t, json.Unmarshal(raw, &decoded)) require.Len(t, decoded.Peers, 2) assert.Equal(t, init.Peers[0], decoded.Peers[0]) assert.Equal(t, init.Peers[1], decoded.Peers[1]) } func TestWorkerInit_PeersOmittedWhenEmpty(t *testing.T) { init := wire.WorkerInit{WorkerID: "w-1", Concurrency: 1} raw, err := json.Marshal(init) require.NoError(t, err) assert.NotContains(t, string(raw), `"peers"`, "empty peers slice must be omitted so legacy workers stay wire-compatible") } // TestRunSelfcheckOnce_ConfiguredPeersNoFreshPeer_NoAlert verifies that // in a multi-worker install a transient local probe failure (or // success) does NOT fire an alert or reset the state machine when // peers are configured but have not produced a fresh observation yet. // The first peer poll lands 2s after startup, and after that on the // peer-poll interval, so the early ticks are a real "lone self vote" // window in production. func TestRunSelfcheckOnce_ConfiguredPeersNoFreshPeer_NoAlert(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) })) defer srv.Close() r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.config.URL = srv.URL creds := &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, } r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: creds, SystemContacts: []wire.SystemContact{ {ID: 1, Kind: "email", Value: "ops@example.com"}, }, // Two peers configured but neither has produced a // fresh observation yet — the selfcheck is the lone // voter. Peers: []wire.PeerInfo{ {WorkerID: "w-2", URL: "http://127.0.0.1:27402"}, {WorkerID: "w-3", URL: "http://127.0.0.1:27403"}, }, }) state := &selfcheckState{consensus: &consensusState{}} r.runSelfcheckOnce(t.Context(), state) assert.Nil(t, state.consensus.downSince, "with configured peers and no fresh peer observations, the down timer must not start") assert.False(t, state.consensus.alertActive, "no alert can fire without a peer-backed down verdict") } // TestRunSelfcheckOnce_ConfiguredPeersWithFreshDownPeer_FiresAfterWait // exercises the happy path: self and one peer both see the master as // down, the consensus verdict is down, and after selfcheckConsensusWait // the alert fires exactly once. func TestRunSelfcheckOnce_ConfiguredPeersWithFreshDownPeer_FiresAfterWait(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) })) defer srv.Close() r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.config.URL = srv.URL r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, SystemContacts: []wire.SystemContact{ {ID: 1, Kind: "email", Value: "ops@example.com"}, }, Peers: []wire.PeerInfo{ {WorkerID: "w-2", URL: "http://127.0.0.1:27402"}, }, }) // Seed a fresh down observation from the peer as if its // selfcheck tick has just completed. r.peerCache.put(peerObservation{ WorkerID: "w-2", Up: false, ObservedAt: time.Now(), }) state := &selfcheckState{consensus: &consensusState{}} // First tick: stamps downSince, does not yet fire. r.runSelfcheckOnce(t.Context(), state) require.NotNil(t, state.consensus.downSince, "peer-backed down verdict must stamp downSince on first tick") assert.False(t, state.consensus.alertActive, "alert must not fire before selfcheckConsensusWait elapses") // Second tick, simulate the 5-minute wait having elapsed // by rewinding downSince past the wait threshold. This // avoids waiting wall-clock time in a unit test. state.consensus.downSince = ptrTime(time.Now().Add(-selfcheckConsensusWait - time.Second)) r.runSelfcheckOnce(t.Context(), state) assert.True(t, state.consensus.alertActive, "down consensus held for selfcheckConsensusWait must fire the alert exactly once") } func TestRunSelfcheckOnce_NonLeaderSkipsDuplicateAlert(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) })) defer srv.Close() r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.config.URL = srv.URL r.applyInit(&wire.WorkerInit{ WorkerID: "worker-local-2", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, SystemContacts: []wire.SystemContact{{ID: 1, Kind: "email", Value: "ops@example.com"}}, Peers: []wire.PeerInfo{{WorkerID: "worker-local-1", URL: "http://127.0.0.1:27401"}}, }) r.peerCache.put(peerObservation{WorkerID: "worker-local-1", Up: false, ObservedAt: time.Now()}) state := &selfcheckState{consensus: &consensusState{}} state.consensus.downSince = ptrTime(time.Now().Add(-selfcheckConsensusWait - time.Second)) r.runSelfcheckOnce(t.Context(), state) assert.True(t, state.consensus.alertActive, "non-leader still marks the incident handled locally") assert.Empty(t, r.RecentNotifications(10), "non-leader must not deliver duplicate system-contact notifications") } func TestFormatSelfcheckLeaderMessage_IncludesOldAndNewLeader(t *testing.T) { r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{WorkerID: "worker-local-2", RegionCode: "local", Concurrency: 1}) out := r.formatSelfcheckLeaderMessage("worker-local-1", "http://localhost:7401/up", consensusDown, 0, 2, 2, 2, 2, 1, "worker-local-2", []string{"worker-local-2", "worker-local-3"}) assert.Contains(t, out, "Изменился мастер воркер оповещений: worker-local-1 -> worker-local-2") assert.Contains(t, out, "Notification leader: worker-local-2") assert.Contains(t, out, "Voters: worker-local-2,worker-local-3") } // TestRunSelfcheckOnce_RecoveryRequiresHeldUpConsensus verifies that // after a peer-backed down alert, recovery only fires after the up // verdict has itself been held for selfcheckConsensusWait, and that // the recovery flow respects the multi-worker gate (an isolated up // tick with no fresh peer must NOT clear the alert). func TestRunSelfcheckOnce_RecoveryRequiresHeldUpConsensus(t *testing.T) { srvOK := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) defer srvOK.Close() r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, SystemContacts: []wire.SystemContact{ {ID: 1, Kind: "email", Value: "ops@example.com"}, }, Peers: []wire.PeerInfo{ {WorkerID: "w-2", URL: "http://127.0.0.1:27402"}, }, }) // Pretend the peer also sees the master as up — the only // way the up verdict is allowed past the multi-worker gate. r.peerCache.put(peerObservation{ WorkerID: "w-2", Up: true, ObservedAt: time.Now(), }) state := &selfcheckState{consensus: &consensusState{}} state.consensus.alertActive = true state.consensus.downSince = ptrTime(time.Now().Add(-20 * time.Minute)) r.config.URL = srvOK.URL // First up tick: stamps the recovery timer; alert stays // active because the up verdict is not yet held for // selfcheckConsensusWait. r.runSelfcheckOnce(t.Context(), state) assert.True(t, state.consensus.alertActive, "recovery alert must not fire on the first up tick") assert.Equal(t, consensusUp, state.consensus.lastVerdict) // Simulate the 5-minute recovery wait having elapsed. state.consensus.downSince = ptrTime(time.Now().Add(-selfcheckConsensusWait - time.Second)) r.runSelfcheckOnce(t.Context(), state) assert.False(t, state.consensus.alertActive, "recovery alert must clear alertActive after selfcheckConsensusWait up consensus") assert.Nil(t, state.consensus.downSince) } // TestRunSelfcheckOnce_RecoveryIsolatedUpDoesNotClearAlert verifies // that an isolated up tick (configured peers, no fresh peer // observation) cannot unilaterally clear an active alert. The // alert must remain active until a peer-backed up verdict is held. func TestRunSelfcheckOnce_RecoveryIsolatedUpDoesNotClearAlert(t *testing.T) { srvOK := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) defer srvOK.Close() r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, SystemContacts: []wire.SystemContact{ {ID: 1, Kind: "email", Value: "ops@example.com"}, }, Peers: []wire.PeerInfo{ {WorkerID: "w-2", URL: "http://127.0.0.1:27402"}, }, }) // Note: no fresh peer observation is seeded — the selfcheck // is the lone voter in this scenario. state := &selfcheckState{consensus: &consensusState{}} state.consensus.alertActive = true state.consensus.downSince = ptrTime(time.Now().Add(-20 * time.Minute)) r.config.URL = srvOK.URL r.runSelfcheckOnce(t.Context(), state) assert.True(t, state.consensus.alertActive, "isolated up tick must not clear an active alert when peers are configured") } // TestRunSelfcheckOnce_NoQuorumDoesNotResetDownTimer verifies that a // NoQuorum verdict neither stamps nor advances the 5-minute down // timer. A brief blip in peer reachability must not page or reset // progress toward the consensus wait. func TestRunSelfcheckOnce_NoQuorumDoesNotResetDownTimer(t *testing.T) { r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{ WorkerID: "w-1", Credentials: &wire.NotificationCredentials{ SMTP: []wire.SMTPCredential{{ID: 1, Name: "primary", Server: "smtp", Port: 25, Login: "a", FromAddress: "a", Password: "x"}}, }, SystemContacts: []wire.SystemContact{ {ID: 1, Kind: "email", Value: "ops@example.com"}, }, Peers: []wire.PeerInfo{ {WorkerID: "w-2", URL: "http://127.0.0.1:27402"}, }, }) // Peer observation is older than peerStatusMaxAge so the // selfcheck is alone in the vote. r.peerCache.put(peerObservation{ WorkerID: "w-2", Up: false, ObservedAt: time.Now().Add(-10 * time.Minute), }) state := &selfcheckState{consensus: &consensusState{}} r.runSelfcheckOnce(t.Context(), state) assert.Nil(t, state.consensus.downSince, "first NoQuorum must not stamp the down timer") assert.False(t, state.consensus.alertActive) // A subsequent NoQuorum must also not advance the timer; // the existing stamp (from a previous verdict) is held. previous := ptrTime(time.Now().Add(-3 * time.Minute)) state.consensus.downSince = previous r.runSelfcheckOnce(t.Context(), state) require.NotNil(t, state.consensus.downSince) assert.Equal(t, *previous, *state.consensus.downSince, "NoQuorum must not advance the down timer") } func TestFormatSelfcheckMessage_IncludesWorkerAndConsensus(t *testing.T) { r := newTestRunner(t, 4, 1, func(interface{}) interface{} { return []wire.CheckResultReport{} }) r.applyInit(&wire.WorkerInit{WorkerID: "worker-local-1", RegionCode: "local", Concurrency: 1}) out := r.formatSelfcheckMessage(selfcheckDownMessage, "http://localhost:7401/up", consensusDown, 0, 3, 3, 2, 2, 2, "worker-local-1", []string{"worker-local-1", "worker-local-2", "worker-local-3"}) assert.Contains(t, out, "Нет связи с основным api") assert.Contains(t, out, "Worker: worker-local-1") assert.Contains(t, out, "Region: local") assert.Contains(t, out, "Target: http://localhost:7401/up") assert.Contains(t, out, "Raft status: lightweight peer quorum") assert.Contains(t, out, "Notification leader: worker-local-1") assert.Contains(t, out, "Consensus: down") assert.Contains(t, out, "Votes: up=0 down=3 total=3 min_votes=2") assert.Contains(t, out, "Workers seen: 3 current votes, 2 fresh peers of 2 configured peers") assert.Contains(t, out, "Voters: worker-local-1,worker-local-2,worker-local-3") assert.Contains(t, out, "Hold time: 5m0s") } func TestConsensusNotificationLeader(t *testing.T) { voters := consensusVoters("worker-local-2", []peerObservation{{WorkerID: "worker-local-3"}, {WorkerID: "worker-local-1"}}) assert.Equal(t, []string{"worker-local-1", "worker-local-2", "worker-local-3"}, voters) assert.Equal(t, "worker-local-1", consensusNotificationLeader(voters)) } func TestConsensusState_NotificationLeaderChanged(t *testing.T) { state := &consensusState{} old, changed := state.notificationLeaderChanged("worker-local-1") assert.False(t, changed) assert.Empty(t, old) old, changed = state.notificationLeaderChanged("worker-local-1") assert.False(t, changed) assert.Empty(t, old) old, changed = state.notificationLeaderChanged("worker-local-2") assert.True(t, changed) assert.Equal(t, "worker-local-1", old) } func ptrTime(t time.Time) *time.Time { return &t }