av
bd5d17b079
- свёртка спрашивает журнал, встречалось ли имя строго раньше по паре (received_at, id), и пишет WARN с атрибутом uncovered_new; повторные молчат. Признак выводится, а не хранится — реестр был бы второй копией факта - добавлена подкоманда `healthlog uncovered`: перечень накопленного, чтение только на чтение, экранированные имена и названные границы носителя - синк документации: ADR о выводе новизны из журнала, две записи в журнал дефектов, два правила промоутом в конвенции, терминал оператора назван адресатом недоверенного входа
396 lines
15 KiB
Go
396 lines
15 KiB
Go
package fold_test
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"log/slog"
|
||
"path/filepath"
|
||
"regexp"
|
||
"strings"
|
||
"testing"
|
||
|
||
"git.vakhrushev.me/av/healthlog/internal/archive"
|
||
"git.vakhrushev.me/av/healthlog/internal/fold"
|
||
"git.vakhrushev.me/av/healthlog/internal/store"
|
||
|
||
_ "modernc.org/sqlite" // прямое подключение к файлу базы — ради теста, ломающего сверку
|
||
)
|
||
|
||
// loggedFold — свёртка с логгером, чьи записи можно прочитать.
|
||
//
|
||
// Отдельный конструктор, потому что событие о новой секции наблюдаемо ТОЛЬКО
|
||
// через лог: в учёте его нет и быть не может — колонка говорит, какие секции в
|
||
// теле есть, а не какая из них встретилась впервые. Путь к файлу базы нужен
|
||
// тесту, ломающему сверку: сломать её изнутри нечем — свёртка держит настоящее
|
||
// хранилище, а заводить интерфейс ради мока значит менять код под тест.
|
||
type loggedFold struct {
|
||
svc *fold.Service
|
||
arch *archive.Archive
|
||
st *store.Store
|
||
log *bytes.Buffer
|
||
dbPath string
|
||
}
|
||
|
||
func newLoggedFold(t *testing.T) loggedFold {
|
||
t.Helper()
|
||
|
||
dir := t.TempDir()
|
||
arch, err := archive.New(filepath.Join(dir, "raw"))
|
||
if err != nil {
|
||
t.Fatalf("архив: %v", err)
|
||
}
|
||
dbPath := filepath.Join(dir, "healthlog.db")
|
||
st, err := store.Open(dbPath)
|
||
if err != nil {
|
||
t.Fatalf("база: %v", err)
|
||
}
|
||
t.Cleanup(func() { _ = st.Close() })
|
||
|
||
var buf bytes.Buffer
|
||
log := slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo}))
|
||
return loggedFold{
|
||
svc: fold.New(arch, st, 0, log),
|
||
arch: arch,
|
||
st: st,
|
||
log: &buf,
|
||
dbPath: dbPath,
|
||
}
|
||
}
|
||
|
||
// breakMerge сносит таблицу объектов: слияние отказывает нетранзиентно, а разбор
|
||
// и запись исхода продолжают работать.
|
||
//
|
||
// В жизни в ту же ветвь ведёт исчерпанный дедлайн свёртки под большим телом
|
||
// (измерено: 63 МиБ держат блокировку 5.019 с) — `store.Transient` дедлайн
|
||
// намеренно не признаёт.
|
||
func breakMerge(t *testing.T, dbPath string) {
|
||
t.Helper()
|
||
|
||
db, err := sql.Open("sqlite", "file:"+dbPath+"?_pragma=busy_timeout(5000)")
|
||
if err != nil {
|
||
t.Fatalf("открытие базы напрямую: %v", err)
|
||
}
|
||
defer func() { _ = db.Close() }()
|
||
|
||
if _, err := db.ExecContext(context.Background(), `DROP TABLE bucket`); err != nil {
|
||
t.Fatalf("снос таблицы объектов: %v", err)
|
||
}
|
||
}
|
||
|
||
// logRecords разбирает захваченные записи лога.
|
||
func logRecords(t *testing.T, buf *bytes.Buffer) []map[string]any {
|
||
t.Helper()
|
||
|
||
var out []map[string]any
|
||
for line := range strings.SplitSeq(strings.TrimSpace(buf.String()), "\n") {
|
||
if line == "" {
|
||
continue
|
||
}
|
||
var rec map[string]any
|
||
if err := json.Unmarshal([]byte(line), &rec); err != nil {
|
||
t.Fatalf("строка лога не JSON: %v", err)
|
||
}
|
||
out = append(out, rec)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// newSections достаёт имена новых секций из записи лога.
|
||
func newSections(t *testing.T, rec map[string]any) []string {
|
||
t.Helper()
|
||
|
||
raw, ok := rec["uncovered_new"]
|
||
if !ok || raw == nil {
|
||
// Пустой список кодировщик пишет как `null` — это и есть «новых имён
|
||
// нет», а не сломанный атрибут.
|
||
return nil
|
||
}
|
||
list, ok := raw.([]any)
|
||
if !ok {
|
||
t.Fatalf("атрибут uncovered_new не список: %#v", raw)
|
||
}
|
||
out := make([]string, 0, len(list))
|
||
for _, v := range list {
|
||
s, ok := v.(string)
|
||
if !ok {
|
||
t.Fatalf("имя секции не строка: %#v", v)
|
||
}
|
||
out = append(out, s)
|
||
}
|
||
return out
|
||
}
|
||
|
||
// withSection дописывает в тело непокрытую секцию с данным именем.
|
||
func withSection(t *testing.T, body []byte, name string) []byte {
|
||
t.Helper()
|
||
|
||
const anchor = `"data": {`
|
||
if !bytes.Contains(body, []byte(anchor)) {
|
||
t.Fatalf("в теле нет объекта data — дописать секцию некуда")
|
||
}
|
||
return bytes.Replace(body, []byte(anchor), []byte(anchor+"\n \""+name+"\": [],"), 1)
|
||
}
|
||
|
||
// Момент появления секции наблюдать больше нечем: телефон дописывает метрики
|
||
// молча, а колонка учёта говорит, какие секции в теле есть, а не какая из них
|
||
// приехала впервые.
|
||
func TestFoldПерваяВстречаСекцииДаётСобытие(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
lf := newLoggedFold(t)
|
||
ctx := context.Background()
|
||
body := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||
|
||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", body)
|
||
stats, err := lf.svc.Fold(ctx, "d1")
|
||
if err != nil {
|
||
t.Fatalf("свёртка: %v", err)
|
||
}
|
||
|
||
if len(stats.UncoveredNew) != 1 || stats.UncoveredNew[0] != "symptoms" {
|
||
t.Fatalf("новых секций %v, ожидалась ровно `symptoms`", stats.UncoveredNew)
|
||
}
|
||
|
||
recs := logRecords(t, lf.log)
|
||
var announced int
|
||
for _, rec := range recs {
|
||
for _, name := range newSections(t, rec) {
|
||
if name != "symptoms" {
|
||
continue
|
||
}
|
||
announced++
|
||
if rec["level"] != "WARN" {
|
||
t.Errorf("уровень записи %v, ожидался WARN: приезд новой секции — «посмотри», а не «разбери сбой»",
|
||
rec["level"])
|
||
}
|
||
}
|
||
}
|
||
if announced != 1 {
|
||
t.Errorf("записей о новой секции %d, ожидалась ровно одна", announced)
|
||
}
|
||
}
|
||
|
||
// Вторая доставка с тем же именем молчит: событие однократно за всю жизнь
|
||
// имени, иначе поток раз в пять минут обесценил бы уровень.
|
||
func TestFoldВтораяВстречаСобытияНеДаёт(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
lf := newLoggedFold(t)
|
||
ctx := context.Background()
|
||
body := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||
|
||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", body)
|
||
if _, err := lf.svc.Fold(ctx, "d1"); err != nil {
|
||
t.Fatalf("первая свёртка: %v", err)
|
||
}
|
||
lf.log.Reset()
|
||
|
||
deliver(t, lf.arch, lf.st, "d2", "Minutes", "auto-1", body)
|
||
stats, err := lf.svc.Fold(ctx, "d2")
|
||
if err != nil {
|
||
t.Fatalf("вторая свёртка: %v", err)
|
||
}
|
||
|
||
if len(stats.UncoveredNew) != 0 {
|
||
t.Errorf("вторая доставка объявила новыми %v", stats.UncoveredNew)
|
||
}
|
||
|
||
recs := logRecords(t, lf.log)
|
||
if len(recs) == 0 {
|
||
t.Fatal("вторая свёртка не записала ничего — чекпоинт молчит вовсе")
|
||
}
|
||
for _, rec := range recs {
|
||
if len(newSections(t, rec)) != 0 {
|
||
t.Errorf("вторая доставка назвала новые секции: %v", rec)
|
||
}
|
||
if rec["msg"] == "delivery folded" && rec["level"] != "INFO" {
|
||
t.Errorf("уровень повторной доставки %v, ожидался INFO", rec["level"])
|
||
}
|
||
}
|
||
|
||
// Перечисление непокрытых секций при этом не меняется: атрибут `uncovered`
|
||
// продолжает называть их все — по нему первую встречу и не отличить.
|
||
if !strings.Contains(lf.log.String(), "symptoms") {
|
||
t.Error("имя секции пропало из записи вовсе: атрибут непокрытых секций не трогается")
|
||
}
|
||
}
|
||
|
||
// Новое имя рядом с уже виденным: событие адресное, а не «в теле что-то новое».
|
||
func TestFoldНовоеИмяРядомСВиденным(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
lf := newLoggedFold(t)
|
||
ctx := context.Background()
|
||
first := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||
second := withSection(t, first, "ecg")
|
||
|
||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", first)
|
||
if _, err := lf.svc.Fold(ctx, "d1"); err != nil {
|
||
t.Fatalf("первая свёртка: %v", err)
|
||
}
|
||
|
||
deliver(t, lf.arch, lf.st, "d2", "Minutes", "auto-1", second)
|
||
stats, err := lf.svc.Fold(ctx, "d2")
|
||
if err != nil {
|
||
t.Fatalf("вторая свёртка: %v", err)
|
||
}
|
||
|
||
if len(stats.UncoveredNew) != 1 || stats.UncoveredNew[0] != "ecg" {
|
||
t.Errorf("новых секций %v, ожидалась ровно `ecg`", stats.UncoveredNew)
|
||
}
|
||
}
|
||
|
||
// Событие однократно за жизнь имени, значит замаскировать его рутинным
|
||
// событием нельзя: перезаписи точек случаются на живом потоке постоянно, а
|
||
// приезд новой секции — один раз.
|
||
func TestFoldНоваяСекцияНеМаскируетсяПерезаписью(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
lf := newLoggedFold(t)
|
||
ctx := context.Background()
|
||
|
||
base := fixture(t, "minute.json")
|
||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", base)
|
||
if _, err := lf.svc.Fold(ctx, "d1"); err != nil {
|
||
t.Fatalf("первая свёртка: %v", err)
|
||
}
|
||
lf.log.Reset()
|
||
|
||
// Те же координаты, другое значение: набор полей совпадает, полнота равна,
|
||
// поэтому побеждает пришедшая — это перезапись.
|
||
changed := regexp.MustCompile(`"qty"\s*:\s*[-0-9.eE+]+`).ReplaceAll(base, []byte(`"qty": 777.5`))
|
||
if bytes.Equal(changed, base) {
|
||
t.Fatal("фикстура не содержит qty — перезапись не устроить")
|
||
}
|
||
deliver(t, lf.arch, lf.st, "d2", "Minutes", "auto-1", withSection(t, changed, "ecg"))
|
||
stats, err := lf.svc.Fold(ctx, "d2")
|
||
if err != nil {
|
||
t.Fatalf("вторая свёртка: %v", err)
|
||
}
|
||
if stats.Overwrites == 0 {
|
||
t.Fatal("перезаписей нет — рутинного события, которое могло бы замаскировать новое, не случилось")
|
||
}
|
||
|
||
var found bool
|
||
for _, rec := range logRecords(t, lf.log) {
|
||
if rec["msg"] != "delivery folded, new uncovered section" {
|
||
continue
|
||
}
|
||
found = true
|
||
if rec["overwrites"] == nil {
|
||
t.Error("счётчик перезаписей исчез из записи: признаки идут атрибутами всегда")
|
||
}
|
||
}
|
||
if !found {
|
||
t.Errorf("запись назвала перезапись, а не новую секцию:\n%s", lf.log.String())
|
||
}
|
||
}
|
||
|
||
// Список непокрытых секций переживает отказ разбора, то есть имя уже записано в
|
||
// учёт. Промолчи здесь — и следующая доставка сочтёт его виденным, а событие не
|
||
// вернётся ничем.
|
||
func TestFoldОтказРазбораНазываетНовуюСекцию(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
lf := newLoggedFold(t)
|
||
ctx := context.Background()
|
||
|
||
// Слой не выводится: плотных метрик в теле нет вовсе.
|
||
body := withSection(t, fixture(t, "sparse_sleep.json"), "cycleTracking")
|
||
deliver(t, lf.arch, lf.st, "d1", "Default", "auto-1", body)
|
||
if _, err := lf.svc.Fold(ctx, "d1"); err == nil {
|
||
t.Fatal("свёртка с неопределимым слоем прошла успешно")
|
||
}
|
||
|
||
var found bool
|
||
for _, rec := range logRecords(t, lf.log) {
|
||
for _, name := range newSections(t, rec) {
|
||
if name == "cycleTracking" {
|
||
found = true
|
||
}
|
||
}
|
||
}
|
||
if !found {
|
||
t.Errorf("запись об отказе не назвала новую секцию:\n%s", lf.log.String())
|
||
}
|
||
|
||
// И имя доехало до учёта — иначе терять было бы нечего.
|
||
d, err := lf.st.LastDelivery(ctx)
|
||
if err != nil {
|
||
t.Fatalf("чтение доставки: %v", err)
|
||
}
|
||
if !strings.Contains(d.UncoveredSections, "cycleTracking") {
|
||
t.Errorf("список в учёте %q не содержит секции", d.UncoveredSections)
|
||
}
|
||
}
|
||
|
||
// Пересборка проигрывает журнал заново, и повторная свёртка обязана дать тот же
|
||
// состав событий: признак новизны — функция журнала, а не числа прогонов.
|
||
func TestFoldПовторнаяСвёрткаДаётТотЖеСоставСобытий(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
lf := newLoggedFold(t)
|
||
ctx := context.Background()
|
||
body := withSection(t, fixture(t, "minute.json"), "symptoms")
|
||
|
||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1", body)
|
||
first, err := lf.svc.Fold(ctx, "d1")
|
||
if err != nil {
|
||
t.Fatalf("первая свёртка: %v", err)
|
||
}
|
||
second, err := lf.svc.Fold(ctx, "d1")
|
||
if err != nil {
|
||
t.Fatalf("повторная свёртка: %v", err)
|
||
}
|
||
|
||
if len(first.UncoveredNew) != len(second.UncoveredNew) {
|
||
t.Fatalf("новых секций %v против %v", first.UncoveredNew, second.UncoveredNew)
|
||
}
|
||
for i := range first.UncoveredNew {
|
||
if first.UncoveredNew[i] != second.UncoveredNew[i] {
|
||
t.Errorf("на месте %d %q против %q", i, first.UncoveredNew[i], second.UncoveredNew[i])
|
||
}
|
||
}
|
||
}
|
||
|
||
// Отказ СЛИЯНИЯ пишет имя в учёт ровно так же, как отказ разбора, — значит и
|
||
// событие терять ему нельзя. Ветвь отдельная, и однажды она уже теряла его:
|
||
// residue собирался вручную и новизну не нёс.
|
||
func TestFoldОтказСлиянияНазываетНовуюСекцию(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
lf := newLoggedFold(t)
|
||
ctx := context.Background()
|
||
|
||
breakMerge(t, lf.dbPath)
|
||
deliver(t, lf.arch, lf.st, "d1", "Minutes", "auto-1",
|
||
withSection(t, fixture(t, "minute.json"), "ecg"))
|
||
if _, err := lf.svc.Fold(ctx, "d1"); err == nil {
|
||
t.Fatal("свёртка без таблицы объектов прошла успешно")
|
||
}
|
||
|
||
var told bool
|
||
for _, rec := range logRecords(t, lf.log) {
|
||
for _, name := range newSections(t, rec) {
|
||
if name == "ecg" {
|
||
told = true
|
||
}
|
||
}
|
||
}
|
||
if !told {
|
||
t.Errorf("запись об отказе слияния не назвала новую секцию:\n%s", lf.log.String())
|
||
}
|
||
|
||
// И событие действительно потеряно быть не могло: имя уже в учёте, то есть
|
||
// следующая доставка сочтёт его виденным.
|
||
d, err := lf.st.LastDelivery(ctx)
|
||
if err != nil {
|
||
t.Fatalf("чтение доставки: %v", err)
|
||
}
|
||
if !strings.Contains(d.UncoveredSections, "ecg") {
|
||
t.Errorf("список в учёте %q не содержит секции — терять было бы нечего", d.UncoveredSections)
|
||
}
|
||
}
|