main_test.go
Rama main · réplica de lectura del repositorio original.
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"path/filepath"
"slices"
"strings"
"testing"
"time"
"github.com/go-telegram/bot/models"
)
// A post that crashes every time must not restart the service forever: after
// maxPostFailures crashes it is skipped and the queue moves on.
func TestProcessSkipsAPostThatKeepsCrashing(t *testing.T) {
db := testStore(t)
ctx := t.Context()
src := channelKey{telegramPlatform, -42}
for _, id := range []string{"1", "2"} {
if err := db.enqueue(ctx, src, id, "message:"+id, []byte(id), time.Now()); err != nil {
t.Fatal(err)
}
}
intakeDone := make(chan struct{})
close(intakeDone)
var alerts, handled []string
alert := func(_ context.Context, text string) error {
alerts = append(alerts, text)
return nil
}
forward := func(p pendingPost, _ *deliveryBook) {
handled = append(handled, p.key)
if p.key == "message:1" {
panic("boom")
}
}
// Each crash stops processing, as a restart follows it.
for i := 1; i <= maxPostFailures; i++ {
err := process(ctx, db, src, intakeDone, alert, forward)
if err == nil || !strings.Contains(err.Error(), "boom") {
t.Fatalf("crash %d: expected an error carrying the panic, got %v", i, err)
}
if count, err := db.failures(ctx, src, "message:1"); err != nil || count != i {
t.Fatalf("crash %d: failures = %d, %v", i, count, err)
}
}
if err := process(ctx, db, src, intakeDone, alert, forward); err != nil {
t.Fatal(err)
}
if want := []string{"message:1", "message:1", "message:1", "message:2"}; !slices.Equal(handled, want) {
t.Errorf("handled %v, want %v", handled, want)
}
if len(alerts) != maxPostFailures+1 || !strings.Contains(alerts[len(alerts)-1], "пропущен") {
t.Errorf("alerts = %q", alerts)
}
if count, err := db.failures(ctx, src, "message:1"); err != nil || count != 0 {
t.Errorf("a skipped post must clear its failures: %d, %v", count, err)
}
if post, err := db.next(ctx, src, time.Now()); err != nil || post.row != 0 {
t.Errorf("queue must be empty: %+v, %v", post, err)
}
}
// A post that crashed once and then went through starts from zero next time.
func TestCompleteClearsFailures(t *testing.T) {
db := testStore(t)
ctx := t.Context()
src := channelKey{maxPlatform, -7}
if err := db.enqueue(ctx, src, "m", "message:m", []byte("m"), time.Now()); err != nil {
t.Fatal(err)
}
if _, err := db.recordFailure(ctx, src, "message:m"); err != nil {
t.Fatal(err)
}
post, err := db.next(ctx, src, time.Now())
if err != nil {
t.Fatal(err)
}
if err := db.complete(ctx, src, post); err != nil {
t.Fatal(err)
}
if count, err := db.failures(ctx, src, "message:m"); err != nil || count != 0 {
t.Errorf("failures = %d, %v", count, err)
}
}
func TestProcessKeepsYoungAlbumQueuedOnShutdown(t *testing.T) {
db := testStore(t)
ctx := t.Context()
src := channelKey{telegramPlatform, -42}
if err := db.enqueue(ctx, src, "1", "album:a", []byte(`{"message_id":1}`), time.Now()); err != nil {
t.Fatal(err)
}
intakeDone := make(chan struct{})
close(intakeDone)
forwarded := false
err := process(ctx, db, src, intakeDone, func(context.Context, string) error { return nil }, func(pendingPost, *deliveryBook) {
forwarded = true
})
if err != nil || forwarded {
t.Fatalf("a young album must wait in the queue: forwarded=%v, err=%v", forwarded, err)
}
p, err := db.next(ctx, src, time.Now().Add(albumWait))
if err != nil || p.key != "album:a" || len(p.payloads) != 1 {
t.Fatalf("album lost on shutdown: %+v, %v", p, err)
}
}
func TestProcessRelaysLateAlbumPartSeparatelyAfterRestart(t *testing.T) {
for _, state := range []string{"done", "partial"} {
t.Run(state, func(t *testing.T) {
ctx := t.Context()
path := filepath.Join(t.TempDir(), "queue.sqlite")
db, err := openStore(ctx, path)
if err != nil {
t.Fatal(err)
}
t.Cleanup(db.close)
firstAttempt, cancel := context.WithCancel(ctx)
defer cancel()
var texts []string
r := &relay{max: &maxClient{baseURL: "https://max.example", publish: &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
var message struct{ Text string }
err := json.NewDecoder(req.Body).Decode(&message)
_ = req.Body.Close()
if err != nil {
return nil, err
}
texts = append(texts, message.Text)
body := io.NopCloser(strings.NewReader(fmt.Sprintf(`{"message":{"body":{"mid":"part.%d"}}}`, len(texts))))
if state == "partial" && len(texts) == 1 {
body = cancelOnClose{body, cancel}
}
return &http.Response{StatusCode: http.StatusOK, Body: body, Header: make(http.Header), Request: req}, nil
}),
}}}
src := &source{channel: &models.ChatFullInfo{ID: -100}, to: []maxChat{{id: -42}}}
source := channelKey{telegramPlatform, -100}
destination := channelKey{maxPlatform, -42}
received := time.Now().Add(-2 * albumWait)
var original [][]byte
for i, text := range []string{strings.Repeat("a", maxTextLimit), "original second part"} {
id := fmt.Sprint(i + 1)
payload := []byte(fmt.Sprintf(`{"message_id":%s,"text":%q}`, id, text))
original = append(original, payload)
if err := db.enqueue(ctx, source, id, "album:a", payload, received); err != nil {
t.Fatal(err)
}
}
book := &deliveryBook{store: db, source: source, post: "album:a"}
p, err := db.next(ctx, source, time.Now())
if err != nil {
t.Fatal(err)
}
r.forwardTelegramPost(firstAttempt, src, p, book)
var saved string
if err := db.QueryRowContext(ctx, `SELECT state FROM deliveries WHERE source = ? AND post = 'album:a' AND destination = ?`, source.String(), destination.String()).Scan(&saved); book.err != nil || err != nil || saved != state {
t.Fatalf("first attempt: state=%q, err=%v, book=%v", saved, err, book.err)
}
if state == "done" {
if err := db.complete(ctx, source, p); err != nil {
t.Fatal(err)
}
}
if err := db.enqueue(ctx, source, "3", "album:a", []byte(`{"message_id":3,"text":"late part"}`), received); err != nil {
t.Fatal(err)
}
db.close()
reopened, err := openStore(ctx, path)
if err != nil {
t.Fatal(err)
}
t.Cleanup(reopened.close)
intakeDone := make(chan struct{})
close(intakeDone)
forward := func(p pendingPost, book *deliveryBook) { r.forwardTelegramPost(ctx, src, p, book) }
for range 2 {
if err := process(ctx, reopened, source, intakeDone, r.alert, forward); err != nil {
t.Fatal(err)
}
}
if len(texts) != 3 || texts[0] != strings.Repeat("a", maxTextLimit) || texts[1] != "original second part" || texts[2] != "late part" {
t.Fatalf("late part lost or original repeated: %q", texts)
}
if err := reopened.QueryRowContext(ctx, `SELECT state FROM deliveries WHERE source = ? AND post = 'album:a#3' AND destination = ?`, source.String(), destination.String()).Scan(&saved); err != nil || saved != "done" {
t.Fatalf("late part needs its own completed delivery: %q, %v", saved, err)
}
payloads, err := reopened.albumPayloads(ctx, source, "album:a")
if err != nil || len(payloads) != 2 || string(payloads[0]) != string(original[0]) || string(payloads[1]) != string(original[1]) {
t.Fatalf("original snapshot changed: %q, %v", payloads, err)
}
})
}
}
func TestForwardTelegramPostKeepsKnownFieldsOnTypeError(t *testing.T) {
f := fakes(t)
db := testStore(t)
p := pendingPost{key: "message:1", payloads: [][]byte{[]byte(`{"message_id":1,"message_thread_id":"unexpected","text":"keep this text"}`)}}
book := &deliveryBook{store: db, source: channelKey{telegramPlatform, -100}, post: p.key}
f.relay.forwardTelegramPost(t.Context(), f.source, p, book)
messages := f.messages()
if book.err != nil || len(messages) != 1 || messages[0].body.Text != "keep this text" {
t.Fatalf("known fields were lost: %+v, %v", messages, book.err)
}
if alerts := f.alerted(); len(alerts) != 1 || !strings.Contains(alerts[0], "разобран не полностью") {
t.Fatalf("partial parse must notify the admin: %q", alerts)
}
}
func TestForwardMaxPostKeepsKnownFieldsOnTypeError(t *testing.T) {
f := fakes(t)
db := testStore(t)
var text string
telegram := &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
var body struct{ Text string }
err := json.NewDecoder(req.Body).Decode(&body)
_ = req.Body.Close()
if err != nil {
return nil, err
}
text = body.Text
return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(`{"ok":true,"result":{"message_id":17}}`)), Header: make(http.Header), Request: req}, nil
})}
f.relay.tgAPI = telegramAPI{baseURL: "https://telegram.example", token: "token", http: telegram}
src := &maxSource{channel: maxChat{id: -42}, to: []*models.ChatFullInfo{{ID: -100}}}
p := pendingPost{key: "message:mid.1", payloads: [][]byte{[]byte(`{"recipient":{"chat_id":-42,"post_id":123},"body":{"mid":"mid.1","text":"keep MAX text"}}`)}}
book := &deliveryBook{store: db, source: channelKey{maxPlatform, -42}, post: p.key}
f.relay.forwardMaxPost(t.Context(), src, p, book)
if book.err != nil || text != "keep MAX text" {
t.Fatalf("known MAX fields were lost: %q, %v", text, book.err)
}
if alerts := f.alerted(); len(alerts) != 1 || !strings.Contains(alerts[0], "разобран не полностью") {
t.Fatalf("partial MAX parse must notify the admin: %q", alerts)
}
}
func TestTelegramMessagesDoesNotKeepMalformedJSON(t *testing.T) {
messages, err := telegramMessages([][]byte{[]byte(`{"message_id":1,"text":"truncated"`), []byte(`{"message_id":2,"text":"valid"}`)})
if err == nil || len(messages) != 1 || messages[0].ID != 2 || messages[0].Text != "valid" {
t.Fatalf("malformed JSON must be skipped without losing valid parts: %+v, %v", messages, err)
}
}
func TestTelegramEditRejectsPartialMessage(t *testing.T) {
f := fakes(t)
db := testStore(t)
source := channelKey{telegramPlatform, -100}
f.source.changes = map[int64]syncOptions{-42: {edits: true}}
seedMirrors(t, db, source, channelKey{maxPlatform, -42}, "message:1", []string{"mid.original"}, mirrorText)
event := change{Kind: telegramEdit, Mid: "1", Group: "message:1", At: 1, Sequence: 1,
Message: json.RawMessage(`{"message_id":1,"message_thread_id":"unexpected","text":"incomplete edit"}`)}
if err := db.enqueueChange(t.Context(), source, event); err != nil {
t.Fatal(err)
}
p, book := nextChange(t, db, source)
f.relay.forwardTelegramPost(t.Context(), f.source, p, book)
if messages := f.messages(); len(messages) != 0 {
t.Fatalf("a partial edit must not overwrite the published post: %+v", messages)
}
if alerts := f.alerted(); len(alerts) != 1 {
t.Fatalf("rejected edit must notify the admin: %q", alerts)
}
}
func TestTelegramLateAlbumEditTouchesOnlyItsOwnCopy(t *testing.T) {
db := testStore(t)
ctx := t.Context()
src := &source{channel: &models.ChatFullInfo{ID: -100}, to: []maxChat{{id: -42}}, changes: map[int64]syncOptions{-42: {edits: true}}}
source := channelKey{telegramPlatform, -100}
destination := channelKey{maxPlatform, -42}
for _, part := range []struct{ key, id, text, mirror string }{
{"album:a", "1", "original album", "mid.original"},
{"album:a#2", "2", "late part", "mid.late"},
} {
payload := []byte(fmt.Sprintf(`{"message_id":%s,"caption":%q}`, part.id, part.text))
if err := db.keepAlbum(ctx, source, part.key, [][]byte{payload}); err != nil {
t.Fatal(err)
}
seedMirrors(t, db, source, destination, part.key, []string{part.mirror}, mirrorCaption)
}
var calls []string
r := &relay{max: &maxClient{baseURL: "https://max.example", publish: &http.Client{
Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
var body struct{ Text string }
err := json.NewDecoder(req.Body).Decode(&body)
_ = req.Body.Close()
if err != nil {
return nil, err
}
calls = append(calls, req.Method+":"+req.URL.Query().Get("message_id")+":"+body.Text)
return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(`{"success":true}`)), Header: make(http.Header), Request: req}, nil
}),
}}}
event := change{Kind: telegramEdit, Mid: "2", Group: "album:a", At: 1, Sequence: 1,
Message: json.RawMessage(`{"message_id":2,"caption":"edited late part"}`)}
if err := db.enqueueChange(ctx, source, event); err != nil {
t.Fatal(err)
}
p, book := nextChange(t, db, source)
r.forwardTelegramPost(ctx, src, p, book)
if book.err != nil || !slices.Equal(calls, []string{"PUT:mid.late:edited late part"}) {
t.Fatalf("late edit must affect only the late copy: %q, %v", calls, book.err)
}
}
func TestTelegramAlbumEditRejectsPartialMessageBeforeSaving(t *testing.T) {
f := fakes(t)
db := testStore(t)
source := channelKey{telegramPlatform, -100}
f.source.changes = map[int64]syncOptions{-42: {edits: true}}
original := []byte(`{"message_id":1,"caption":"original caption"}`)
if err := db.keepAlbum(t.Context(), source, "album:a", [][]byte{original}); err != nil {
t.Fatal(err)
}
seedMirrors(t, db, source, channelKey{maxPlatform, -42}, "album:a", []string{"mid.original"}, mirrorCaption)
event := change{Kind: telegramEdit, Mid: "1", Group: "album:a", At: 1, Sequence: 1,
Message: json.RawMessage(`{"message_id":1,"message_thread_id":"unexpected","caption":"incomplete edit"}`)}
if err := db.enqueueChange(t.Context(), source, event); err != nil {
t.Fatal(err)
}
p, book := nextChange(t, db, source)
f.relay.forwardTelegramPost(t.Context(), f.source, p, book)
if messages := f.messages(); len(messages) != 0 {
t.Fatalf("a partial edit must not overwrite the published album: %+v", messages)
}
payloads, err := db.albumPayloads(t.Context(), source, "album:a")
if err != nil || len(payloads) != 1 || string(payloads[0]) != string(original) {
t.Fatalf("a partial edit must not replace the saved album: %q, %v", payloads, err)
}
if alerts := f.alerted(); len(alerts) != 1 {
t.Fatalf("rejected edit must notify the admin: %q", alerts)
}
}
func TestForwardMalformedPostWarnsThatItWasSkipped(t *testing.T) {
for _, platform := range []platform{telegramPlatform, maxPlatform} {
t.Run(string(platform), func(t *testing.T) {
f := fakes(t)
p := pendingPost{key: "message:1", payloads: [][]byte{[]byte(`{"body":`)}}
book := &deliveryBook{store: testStore(t), source: channelKey{platform, -100}, post: p.key}
if platform == telegramPlatform {
f.relay.forwardTelegramPost(t.Context(), f.source, p, book)
} else {
f.relay.forwardMaxPost(t.Context(), &maxSource{}, p, book)
}
if alerts := f.alerted(); len(alerts) != 1 || !strings.Contains(alerts[0], "пост пропущен") || strings.Contains(alerts[0], "переносим") {
t.Fatalf("malformed post needs an accurate warning: %q", alerts)
}
})
}
}
func TestProcessCancellationWhileCompletingSkippedPostIsClean(t *testing.T) {
db := testStore(t)
source := channelKey{telegramPlatform, -100}
if err := db.enqueue(t.Context(), source, "1", "message:1", []byte(`{"message_id":1}`), time.Now()); err != nil {
t.Fatal(err)
}
for range maxPostFailures {
if _, err := db.recordFailure(t.Context(), source, "message:1"); err != nil {
t.Fatal(err)
}
}
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
intakeDone := make(chan struct{})
close(intakeDone)
err := process(ctx, db, source, intakeDone, func(context.Context, string) error {
cancel()
return nil
}, func(pendingPost, *deliveryBook) { t.Error("a repeatedly crashing post must be skipped") })
if err != nil {
t.Fatalf("shutdown during completion must be clean: %v", err)
}
p, err := db.next(t.Context(), source, time.Now())
if err != nil || p.key != "message:1" {
t.Fatalf("unfinished cleanup must remain queued: %+v, %v", p, err)
}
}
func TestProcessDoesNotHideDeliveryStorageErrorOnCancellation(t *testing.T) {
db := testStore(t)
source := channelKey{telegramPlatform, -100}
if err := db.enqueue(t.Context(), source, "1", "message:1", []byte(`{"message_id":1}`), time.Now()); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
want := errors.New("failed to save publication state")
err := process(ctx, db, source, make(chan struct{}), func(context.Context, string) error { return nil }, func(_ pendingPost, book *deliveryBook) {
book.err = want
cancel()
})
if !errors.Is(err, want) {
t.Fatalf("shutdown must not hide an unrelated storage failure: %v", err)
}
}