av/transcriber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

удалён вход Telegram, владелец записи стал обязателен в схеме

Browse Source 
- убраны клиент бота, транспорт обновлений, отправитель сообщений, сборка
  входа при старте, секция настроек и зависимость go-telegram-bot-api; из
  конвейера ушла доставка ответа отправителю — исход виден опросом готовности.
  Колонки адресата и значение источника остались в схеме: применённые шаги не
  переписываются
- шаг 202608140003 запрещает пустого владельца у аудиозаписи и у файла;
  существующие строки он не проверяет, и это принято сознательно — искать их
  надо запросом до выкладки
- ревью нашло два пред-существующих дефекта, оба закрыты: пустой второй ответ
  распознавателя стирал сохранённую расшифровку, а пустая расшифровка перестала
  быть заметной вместе с убранной доставкой. Попутно поднят golang.org/x/image
  до v0.45.0 — красный шаг vulns, воспроизводился и на чистом master
This commit is contained in:
av
2026-08-15 07:24:35 +03:00
parent 97ceb7bb69
commit 8f7c3a057a
74 changed files with 2453 additions and 2733 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/service/pipeline_test.go
+74 -52
View File
@@ -60,32 +60,12 @@ func (m *failingMetaViewer) GetInfo(context.Context, string) (*contract.AudioInf
return nil, errors.New("запись не читается")
}
// recordingSender запоминает, что отправлено.
type recordingSender struct {
mu sync.Mutex
messages []string
}
func (s *recordingSender) Send(text string, chatId int64, replyMsgId *int) error {
s.mu.Lock()
defer s.mu.Unlock()
s.messages = append(s.messages, text)
return nil
}
func (s *recordingSender) sent() []string {
s.mu.Lock()
defer s.mu.Unlock()
return append([]string(nil), s.messages...)
}
type pipelineEnv struct {
app core.App
service *TranscribeService
repos Repositories
recordRepo *pbrepo.AudioRecordRepository
fileRepo *pbrepo.FileRepository
sender *recordingSender
}
// testLimits — пределы простоя проверок. Числа боевые; проверка застревания
@@ -104,6 +84,19 @@ func newPipelineEnvWith(
rec contract.AudioRecognizer,
) *pipelineEnv {
t.Helper()
return newPipelineEnvWithLogger(t, metaviewer, converter, rec, slog.New(slog.DiscardHandler))
}
// newPipelineEnvWithLogger — то же с подменённым журналом: проверке, судящей
// строку журнала, нужен свой, а не общий.
func newPipelineEnvWithLogger(
t *testing.T,
metaviewer contract.AudioMetaViewer,
converter contract.AudioFileConverter,
rec contract.AudioRecognizer,
logger *slog.Logger,
) *pipelineEnv {
t.Helper()
app, err := pbrepo.New(t.TempDir())
require.NoError(t, err)
@@ -128,9 +121,7 @@ func newPipelineEnvWith(
Recognitions: pbrepo.NewRecognitionRepository(app),
Events: pbrepo.NewRecordEventRepository(app),
}
sender := &recordingSender{}
svc := NewTranscribeService(repos, metaviewer, converter, rec, sender, testLimits, slog.New(slog.DiscardHandler))
svc := NewTranscribeService(repos, metaviewer, converter, rec, testLimits, logger)
return &pipelineEnv{
app: app,
@@ -138,15 +129,16 @@ func newPipelineEnvWith(
repos: repos,
recordRepo: recordRepo,
fileRepo: fileRepo,
sender: sender,
}
}
// newTelegramRecord заводит запись — так, как её завёл бы приём из бота.
func newTelegramRecord(t *testing.T, env *pipelineEnv) *entity.AudioRecord {
// newRecord заводит запись — так, как её заводит приём по HTTP: от имени
// вошедшего, потому что ничьей записи в хранилище не бывает.
func newRecord(t *testing.T, env *pipelineEnv) *entity.AudioRecord {
t.Helper()
record, err := env.service.CreateJobFromTelegram(t.Context(), strings.NewReader("запись"), "voice.ogg", 100, 1)
record, err := env.service.CreateJobFromApi(
t.Context(), strings.NewReader("запись"), "voice.ogg", newOwner(t, env.app))
require.NoError(t, err)
return record
}
@@ -175,6 +167,17 @@ func enteredStateAt(t *testing.T, env *pipelineEnv, recordID string, moment time
require.NoError(t, env.app.Save(record))
}
// setAttempts ставит записи число отказов: так она выглядит, когда шаг отказал
// столько раз подряд, что предел исчерпан.
func setAttempts(t *testing.T, env *pipelineEnv, recordID string, attempts int) {
t.Helper()
record, err := env.app.FindRecordById(migrations.RecordsCollection, recordID)
require.NoError(t, err)
record.Set("attempts", attempts)
require.NoError(t, env.app.Save(record))
}
// drain крутит конвейер, пока он двигает записи. Паузы опроса снимаются: они
// проверяются отдельно, а здесь мешают дойти до конца.
func drain(t *testing.T, env *pipelineEnv, recordID string) {
@@ -217,7 +220,7 @@ func readRecord(t *testing.T, env *pipelineEnv, id string) *entity.AudioRecord {
func TestRecordHaltsAfterAttemptLimit(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &failingConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
// Захват без выполнения шага — так это выглядит при гибели процесса: отказа
// шаг объявить не успевает, а попытка засчитана.
@@ -239,9 +242,6 @@ func TestRecordHaltsAfterAttemptLimit(t *testing.T) {
assert.Equal(t, entity.StateUploaded, after.State, "рубеж пережил остановку")
assert.Greater(t, after.Attempts, maxAttempts, "число отказов сохранено")
require.Len(t, env.sender.sent(), 1, "отправитель узнал о неудаче")
assert.Contains(t, env.sender.sent()[0], "попытки исчерпаны")
// И из выборки она исчезла.
_, err = env.recordRepo.FindAndAcquire(entity.WorkingStages())
var missing *contract.JobNotFoundError
@@ -265,7 +265,7 @@ func expireAcquisition(t *testing.T, env *pipelineEnv, recordID string) {
func TestHaltedRecordResumesFromItsStage(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
// Доводим до рубежа приведения и останавливаем на нём.
require.NoError(t, env.service.RunStep(t.Context()))
@@ -302,7 +302,7 @@ func TestHaltedRecordResumesFromItsStage(t *testing.T) {
func TestBothFileLinksSurvivePipeline(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
drain(t, env, record.Id)
after := readRecord(t, env, record.Id)
@@ -327,7 +327,7 @@ func TestBothFileLinksSurvivePipeline(t *testing.T) {
func TestOutcomeDoesNotDependOnWorkerCount(t *testing.T) {
for _, workers := range []int{1, 4} {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
for range 20 {
clearDelay(t, env, record.Id)
@@ -361,7 +361,7 @@ func TestOutcomeDoesNotDependOnWorkerCount(t *testing.T) {
// Записи принимаются и не двигаются, и это режим, а не поломка.
func TestZeroWorkersLeaveRecordUntouched(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
// Пул нулевого размера ни одного прогона не делает — потому запись остаётся
// там, где её оставил приём.
@@ -393,7 +393,7 @@ func TestPostponeKeepsStuckCountdown(t *testing.T) {
func TestStuckRecordIsHalted(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
enteredStateAt(t, env, record.Id, time.Now().Add(-2*testLimits.Own))
err := env.service.RunStep(t.Context())
@@ -406,8 +406,6 @@ func TestStuckRecordIsHalted(t *testing.T) {
assert.Equal(t, entity.HaltReasonStuck, *after.HaltReason, "причина названа")
assert.Equal(t, entity.StateUploaded, after.State, "рубеж сохранён")
require.Len(t, env.sender.sent(), 1, "отправитель узнал о неудаче")
assert.Contains(t, env.sender.sent()[0], "застряла")
}
// Конечный рубеж под предел простоя не подпадает: стоять в нём запись будет
@@ -415,7 +413,7 @@ func TestStuckRecordIsHalted(t *testing.T) {
func TestDoneRecordIsNeverStuck(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
drain(t, env, record.Id)
enteredStateAt(t, env, record.Id, time.Now().Add(-10*24*time.Hour))
@@ -434,7 +432,7 @@ func TestDoneRecordIsNeverStuck(t *testing.T) {
func TestOnlyHolderWritesResult(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
first, err := env.recordRepo.FindAndAcquire(entity.WorkingStages())
require.NoError(t, err)
@@ -457,12 +455,12 @@ func TestOnlyHolderWritesResult(t *testing.T) {
after := readRecord(t, env, record.Id)
assert.Equal(t, entity.StateUploaded, after.State, "рубеж не сдвинут потерявшим захват")
assert.Empty(t, env.sender.sent(), "и отправителю от него ничего не ушло")
}
// Критерий приёмки 7. Всякий способ вывести запись из работы сообщает
// отправителю: причин остановки больше одной, и обязанность у них общая.
func TestEveryHaltReasonNotifiesSender(t *testing.T) {
// Критерий приёмки 7. Всякий способ вывести запись из работы оставляет причину
// остановки: причин больше одной, и обязанность у них общая. Отправитель узнаёт
// исход опросом готовности, а владелец сервиса — журналом событий записи.
func TestEveryHaltReasonRecordsItsCause(t *testing.T) {
reasons := []struct {
name string
halt func(t *testing.T, env *pipelineEnv, recordID string)
@@ -481,18 +479,40 @@ func TestEveryHaltReasonNotifiesSender(t *testing.T) {
require.ErrorAs(t, env.service.RunStep(t.Context()), &noop)
},
},
{
name: "исчерпанные отказы",
halt: func(t *testing.T, env *pipelineEnv, recordID string) {
setAttempts(t, env, recordID, maxAttempts+1)
var noop *contract.NoopJobError
require.ErrorAs(t, env.service.RunStep(t.Context()), &noop)
},
},
}
for _, reason := range reasons {
t.Run(reason.name, func(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &failingConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
reason.halt(t, env, record.Id)
after := readRecord(t, env, record.Id)
require.True(t, after.IsHalted(), "запись остановлена")
require.Len(t, env.sender.sent(), 1, "отправитель узнал о неудаче")
// Признак остановки и причина ставятся одним движением, поэтому
// вторым утверждением берётся **журнал событий**: он пишется
// отдельной строкой, отдельным сохранением, и упасть может сам по
// себе. Прежде эту роль играл счёт ответов отправителю; ответы ушли
// вместе с входом Telegram, и без замены проверка вывелась бы из
// собственной предыдущей строки.
events := recordEvents(t, env, record.Id)
require.Len(t, events, 1, "остановка оставила строку журнала событий")
outcome := events[0].GetString("outcome")
assert.Contains(t,
[]string{entity.EventOutcomeHalted, entity.EventOutcomeFailed}, outcome,
"строка журнала называет исход остановкой")
assert.NotEmpty(t, events[0].GetString("outcome_text"), "и несёт причину")
})
}
}
@@ -503,7 +523,7 @@ func TestEveryHaltReasonNotifiesSender(t *testing.T) {
func TestRepeatedStoreKeepsSingleText(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &okConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
first, err := env.repos.Texts.Put(record.Id, entity.TextKindTranscript, "первый разбор")
require.NoError(t, err)
@@ -535,7 +555,7 @@ func TestRetryDelayGrowsAndCaps(t *testing.T) {
func TestFailedStepSchedulesRetryWithGrowingDelay(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &failingConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
// Ссылка переставляется на запись о файле без содержимого: шаг отказывает на
// получении рабочей копии — то есть отказом, а не приговором записи.
@@ -544,6 +564,8 @@ func TestFailedStepSchedulesRetryWithGrowingDelay(t *testing.T) {
empty := core.NewRecord(files)
empty.Set("location", entity.LocationLocal)
empty.Set("size", 1)
// Владелец обязателен и у файла: схема ничьих не принимает.
empty.Set("owner", record.OwnerID)
require.NoError(t, env.app.Save(empty))
stored, err := env.app.FindRecordById(migrations.RecordsCollection, record.Id)
@@ -606,7 +628,7 @@ func TestWorkFilesRemovedAfterConversionFailure(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &failingConverter{})
newTelegramRecord(t, env)
newRecord(t, env)
leftovers, err := filepath.Glob(filepath.Join(tempDir, "transcriber-*"))
require.NoError(t, err)
@@ -624,7 +646,7 @@ func TestWorkFilesRemovedAfterConversionFailure(t *testing.T) {
func TestRecordNeverPointsToMissingFile(t *testing.T) {
env := newPipelineEnv(t, &okMetaViewer{}, &failingConverter{})
record := newTelegramRecord(t, env)
record := newRecord(t, env)
require.NoError(t, env.service.RunStep(t.Context()))
@@ -714,7 +736,7 @@ func TestHaltIsCountedAsFailureAndLoggedOnce(t *testing.T) {
beforeOk := stageCount(t, entity.StateUploaded, "false")
beforeErr := stageCount(t, entity.StateUploaded, "true")
record := newTelegramRecord(t, env)
record := newRecord(t, env)
require.NoError(t, env.service.RunStep(t.Context()))
after := readRecord(t, env, record.Id)
@@ -739,7 +761,7 @@ func TestPostponeWritesNoEvent(t *testing.T) {
rec.inProgress.Store(5)
env := newPipelineEnvWith(t, &okMetaViewer{}, &okConverter{}, rec)
record := newTelegramRecord(t, env)
record := newRecord(t, env)
require.NoError(t, env.service.RunStep(t.Context())) // приведение
require.NoError(t, env.service.RunStep(t.Context())) // отправка
require.Equal(t, entity.StateSubmitted, readRecord(t, env, record.Id).State)