mirror of
https://github.com/rclone/rclone.git
synced 2026-11-07 18:22:55 -05:00
Nick Craig-Wood
846836adc0
This adds core/events which streams events over a single HTTP connection as newline delimited JSON, one object per line, each with a type. Log entries are sent as they are made with type "log" in the same format as core/log returns, so they can be attributed to the rc job which made them. This doesn't need the log buffer to be enabled as the events aren't stored, but if it is enabled then the since parameter sends the entries from the buffer first so a client which reconnects doesn't miss any. The sequence numbers and the JSON for the log entries are now made once for the log buffer and the subscribers to share.
472 lines
16 KiB
Go
472 lines
16 KiB
Go
package log
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"math"
|
|
"sync"
|
|
"testing"
|
|
|
|
"github.com/rclone/rclone/fs"
|
|
"github.com/rclone/rclone/fs/rc"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
)
|
|
|
|
type testAttributionKey struct{}
|
|
|
|
// testAttributionCtx is attributed to rc job 1 in stats group "job/1"
|
|
var testAttributionCtx = context.WithValue(context.Background(), testAttributionKey{}, Attribution{JobID: 1, Group: "job/1"})
|
|
|
|
func init() {
|
|
// Attribute logs in these tests from the context value above,
|
|
// in place of the fs/accounting and fs/rc/jobs implementations.
|
|
fs.StatsGroupFromContext = func(ctx context.Context) (string, bool) {
|
|
attribution, _ := ctx.Value(testAttributionKey{}).(Attribution)
|
|
return attribution.Group, attribution.Group != ""
|
|
}
|
|
fs.JobIDFromContext = func(ctx context.Context) (int64, bool) {
|
|
attribution, _ := ctx.Value(testAttributionKey{}).(Attribution)
|
|
return attribution.JobID, attribution.JobID != 0
|
|
}
|
|
}
|
|
|
|
// testEntry is a decoded log entry from the Buffer.
|
|
type testEntry struct {
|
|
Seq int64 `json:"seq"`
|
|
Level string `json:"level"`
|
|
Msg string `json:"msg"`
|
|
}
|
|
|
|
// decodeEntries decodes entries checking they are all valid JSON.
|
|
func decodeEntries(t *testing.T, entries Entries) (out []testEntry) {
|
|
t.Helper()
|
|
for _, entry := range entries {
|
|
var e testEntry
|
|
require.NoError(t, json.Unmarshal(entry, &e), string(entry))
|
|
out = append(out, e)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// addEntry adds a log entry in the same format as the JSON log to b.
|
|
func addEntry(b *Buffer, level slog.Level, msg string) {
|
|
add(b, level, Attribution{}, fmt.Sprintf(`{"level":%q,"msg":%q}`+"\n", slogLevelToString(level), msg))
|
|
}
|
|
|
|
// add adds a log entry to b the way the log output does.
|
|
func add(b *Buffer, level slog.Level, attribution Attribution, text string) {
|
|
seq := nextSeq.Add(1) - 1
|
|
b.add(seq, level, attribution, makeEntry(seq, text))
|
|
}
|
|
|
|
// newTestBuffer makes an empty buffer of size with sequence numbers starting from 0.
|
|
func newTestBuffer(size int64) *Buffer {
|
|
nextSeq.Store(0)
|
|
b := &Buffer{}
|
|
b.SetSize(size)
|
|
return b
|
|
}
|
|
|
|
func TestBufferDisabled(t *testing.T) {
|
|
b := newTestBuffer(0)
|
|
assert.False(t, b.Enabled())
|
|
addEntry(b, slog.LevelInfo, "dropped")
|
|
// The entry used up a sequence number but wasn't stored
|
|
assert.Equal(t, int64(1), b.Seq())
|
|
entries, next, lost := b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Equal(t, Entries{}, entries)
|
|
assert.Equal(t, int64(1), next)
|
|
assert.Equal(t, int64(1), lost)
|
|
}
|
|
|
|
func TestBufferGet(t *testing.T) {
|
|
b := newTestBuffer(1024)
|
|
assert.True(t, b.Enabled())
|
|
for i := range 10 {
|
|
level := slog.LevelInfo
|
|
if i%2 == 1 {
|
|
level = slog.LevelError
|
|
}
|
|
addEntry(b, level, fmt.Sprintf("msg%d", i))
|
|
}
|
|
assert.Equal(t, int64(10), b.Seq())
|
|
|
|
for _, test := range []struct {
|
|
name string
|
|
from, to int64
|
|
level slog.Level
|
|
limit int
|
|
wantSeqs []int64
|
|
wantNext int64
|
|
wantLost int64
|
|
}{
|
|
{name: "all", from: 0, to: math.MaxInt64, level: slog.LevelDebug, wantSeqs: []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, wantNext: 10},
|
|
{name: "range", from: 3, to: 6, level: slog.LevelDebug, wantSeqs: []int64{3, 4, 5}, wantNext: 6},
|
|
{name: "level", from: 0, to: math.MaxInt64, level: slog.LevelError, wantSeqs: []int64{1, 3, 5, 7, 9}, wantNext: 10},
|
|
{name: "limit", from: 2, to: math.MaxInt64, level: slog.LevelDebug, limit: 3, wantSeqs: []int64{2, 3, 4}, wantNext: 5},
|
|
{name: "limit and level", from: 0, to: math.MaxInt64, level: slog.LevelError, limit: 2, wantSeqs: []int64{1, 3}, wantNext: 4},
|
|
{name: "negative from", from: -5, to: 2, level: slog.LevelDebug, wantSeqs: []int64{0, 1}, wantNext: 2},
|
|
{name: "from the end", from: 10, to: math.MaxInt64, level: slog.LevelDebug, wantNext: 10},
|
|
{name: "from past the end", from: 100, to: math.MaxInt64, level: slog.LevelDebug, wantNext: 10},
|
|
{name: "to before from", from: 5, to: 2, level: slog.LevelDebug, wantNext: 5},
|
|
} {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
entries, next, lost := b.Get(test.from, test.to, Filter{Level: test.level}, test.limit)
|
|
var gotSeqs []int64
|
|
for _, e := range decodeEntries(t, entries) {
|
|
gotSeqs = append(gotSeqs, e.Seq)
|
|
assert.Equal(t, fmt.Sprintf("msg%d", e.Seq), e.Msg)
|
|
}
|
|
assert.Equal(t, test.wantSeqs, gotSeqs)
|
|
assert.Equal(t, test.wantNext, next)
|
|
assert.Equal(t, test.wantLost, lost)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestBufferDropsOldEntries(t *testing.T) {
|
|
b := newTestBuffer(1024)
|
|
for i := range 100 {
|
|
addEntry(b, slog.LevelInfo, fmt.Sprintf("message number %d", i))
|
|
}
|
|
assert.LessOrEqual(t, b.size, int64(1024))
|
|
|
|
entries, next, lost := b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
decoded := decodeEntries(t, entries)
|
|
require.NotEmpty(t, decoded)
|
|
assert.Greater(t, lost, int64(0))
|
|
assert.Equal(t, lost, decoded[0].Seq, "lost should be the number of entries dropped")
|
|
assert.Equal(t, int64(100), next)
|
|
assert.Equal(t, int64(99), decoded[len(decoded)-1].Seq)
|
|
assert.Equal(t, int(100-lost), len(decoded))
|
|
|
|
// The window is entirely in the dropped entries
|
|
entries, next, gotLost := b.Get(1, 3, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Empty(t, entries)
|
|
assert.Equal(t, int64(2), gotLost)
|
|
assert.Equal(t, int64(3), next)
|
|
|
|
// An entry bigger than the Buffer uses a sequence number but
|
|
// isn't stored and doesn't evict the other entries
|
|
addEntry(b, slog.LevelInfo, string(make([]byte, 2048)))
|
|
assert.Equal(t, int64(101), b.Seq())
|
|
entries, next, lost = b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Equal(t, len(decoded), len(entries))
|
|
assert.Equal(t, int64(101), next)
|
|
assert.Equal(t, decoded[0].Seq+1, lost, "dropped entries plus the oversized one")
|
|
entries, next, lost = b.Get(100, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Empty(t, entries)
|
|
assert.Equal(t, int64(101), next)
|
|
assert.Equal(t, int64(1), lost)
|
|
|
|
// A gap in the middle of the range is counted as lost too
|
|
addEntry(b, slog.LevelInfo, "after the gap")
|
|
entries, next, lost = b.Get(99, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Len(t, entries, 2)
|
|
assert.Equal(t, int64(102), next)
|
|
assert.Equal(t, int64(1), lost)
|
|
|
|
// Limit stops before the gap
|
|
entries, next, lost = b.Get(99, math.MaxInt64, Filter{Level: slog.LevelDebug}, 1)
|
|
assert.Len(t, entries, 1)
|
|
assert.Equal(t, int64(101), next)
|
|
assert.Equal(t, int64(1), lost)
|
|
|
|
// Disabling discards the entries
|
|
addEntry(b, slog.LevelInfo, "kept")
|
|
b.SetSize(0)
|
|
assert.False(t, b.Enabled())
|
|
entries, _, _ = b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Empty(t, entries)
|
|
assert.Equal(t, int64(103), b.Seq(), "sequence numbers carry on")
|
|
}
|
|
|
|
// Get shouldn't scan the whole buffer when the filter matches very little
|
|
func TestBufferGetMaxExamine(t *testing.T) {
|
|
b := newTestBuffer(int64(16 * fs.Mebi))
|
|
for i := range maxExamine + 10 {
|
|
add(b, slog.LevelInfo, Attribution{JobID: 2}, fmt.Sprintf(`{"msg":"msg%d"}`, i))
|
|
}
|
|
add(b, slog.LevelInfo, Attribution{JobID: 1}, `{"msg":"wanted"}`)
|
|
|
|
// Nothing matches in the first maxExamine entries so Get stops there
|
|
entries, next, lost := b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug, JobID: 1}, 10)
|
|
assert.Empty(t, entries)
|
|
assert.Equal(t, int64(maxExamine), next)
|
|
assert.Equal(t, int64(0), lost)
|
|
|
|
// Carrying on from next finds it
|
|
entries, next, _ = b.Get(next, math.MaxInt64, Filter{Level: slog.LevelDebug, JobID: 1}, 10)
|
|
decoded := decodeEntries(t, entries)
|
|
require.Len(t, decoded, 1)
|
|
assert.Equal(t, "wanted", decoded[0].Msg)
|
|
assert.Equal(t, int64(maxExamine+11), next)
|
|
|
|
// With no limit everything is examined
|
|
entries, _, _ = b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug, JobID: 1}, 0)
|
|
assert.Len(t, entries, 1)
|
|
}
|
|
|
|
func TestBufferInvalidJSON(t *testing.T) {
|
|
b := newTestBuffer(1024)
|
|
for _, text := range []string{
|
|
"not \"JSON\"\n",
|
|
"control chars \a\x00\xff\U0001F600",
|
|
"{",
|
|
"{ }",
|
|
"{}",
|
|
} {
|
|
add(b, slog.LevelInfo, Attribution{}, text)
|
|
}
|
|
entries, _, _ := b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
decoded := decodeEntries(t, entries)
|
|
require.Len(t, decoded, 5)
|
|
assert.Equal(t, `not "JSON"`, decoded[0].Msg)
|
|
assert.Equal(t, "control chars \a\x00\ufffd\U0001F600", decoded[1].Msg)
|
|
assert.Equal(t, "{", decoded[2].Msg)
|
|
assert.Equal(t, "{ }", decoded[3].Msg)
|
|
assert.Equal(t, "{}", decoded[4].Msg)
|
|
// Check they can all be marshalled
|
|
_, err := json.Marshal(entries)
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func TestEntriesString(t *testing.T) {
|
|
entries := Entries{json.RawMessage(`{"seq":0}`), json.RawMessage(`{"seq":1}`)}
|
|
assert.Equal(t, "2 log entries", entries.String())
|
|
// Check that the entries aren't rendered when logging an rc reply
|
|
assert.Equal(t, "map[_logs:map[entries:2 log entries]]", fmt.Sprintf("%+v", rc.Params{"_logs": rc.Params{"entries": entries}}))
|
|
// but that they marshal as JSON
|
|
out, err := json.Marshal(entries)
|
|
require.NoError(t, err)
|
|
assert.Equal(t, `[{"seq":0},{"seq":1}]`, string(out))
|
|
}
|
|
|
|
// Check setting the option connects and disconnects the buffer
|
|
func TestSetBufferSize(t *testing.T) {
|
|
oldOpt := Opt.BufferSize
|
|
defer func() {
|
|
Opt.BufferSize = oldOpt
|
|
setBufferSize()
|
|
}()
|
|
outputs := func() int {
|
|
Handler.mu.Lock()
|
|
defer Handler.mu.Unlock()
|
|
return len(Handler.outputExtra)
|
|
}
|
|
Opt.BufferSize = 0
|
|
setBufferSize()
|
|
assert.False(t, Recent.Enabled())
|
|
start := outputs()
|
|
fs.Errorf(nil, "not buffered")
|
|
|
|
Opt.BufferSize = 64 * 1024
|
|
setBufferSize()
|
|
assert.True(t, Recent.Enabled())
|
|
assert.Equal(t, start+1, outputs())
|
|
seq := Recent.Seq()
|
|
fs.Errorf(nil, "buffered")
|
|
entries, _, _ := Recent.Get(seq, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
decoded := decodeEntries(t, entries)
|
|
require.Len(t, decoded, 1)
|
|
assert.Equal(t, "buffered", decoded[0].Msg)
|
|
|
|
// Setting again shouldn't connect again
|
|
setBufferSize()
|
|
assert.Equal(t, start+1, outputs())
|
|
|
|
Opt.BufferSize = 0
|
|
setBufferSize()
|
|
assert.False(t, Recent.Enabled())
|
|
assert.Equal(t, start, outputs())
|
|
fs.Errorf(nil, "not buffered")
|
|
entries, _, _ = Recent.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Empty(t, entries)
|
|
}
|
|
|
|
// Check the Buffer works as an output of the handler.
|
|
func TestBufferFromHandler(t *testing.T) {
|
|
b := newTestBuffer(64 * 1024)
|
|
h := NewOutputHandler(io.Discard, nil, logFormatDate|logFormatTime)
|
|
h.addRecordOutput(true, func(r slog.Record, attribution Attribution, text string) {
|
|
seq := nextSeq.Add(1) - 1
|
|
b.add(seq, r.Level, attribution, makeEntry(seq, text))
|
|
})
|
|
|
|
r := slog.NewRecord(t0, fs.SlogLevelNotice, "hello", 0)
|
|
r.AddAttrs(slog.String("object", "file.txt"))
|
|
require.NoError(t, h.Handle(testAttributionCtx, r))
|
|
require.NoError(t, h.Handle(context.Background(), slog.NewRecord(t0, slog.LevelError, "oops", 0)))
|
|
|
|
entries, next, lost := b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug}, 0)
|
|
assert.Equal(t, int64(2), next)
|
|
assert.Equal(t, int64(0), lost)
|
|
decoded := decodeEntries(t, entries)
|
|
require.Len(t, decoded, 2)
|
|
assert.Equal(t, testEntry{Seq: 0, Level: "notice", Msg: "hello"}, decoded[0])
|
|
assert.Equal(t, testEntry{Seq: 1, Level: "error", Msg: "oops"}, decoded[1])
|
|
|
|
var full map[string]any
|
|
require.NoError(t, json.Unmarshal(entries[0], &full))
|
|
assert.Equal(t, "file.txt", full["object"])
|
|
assert.Equal(t, "job/1", full["group"])
|
|
assert.Equal(t, float64(1), full["jobid"])
|
|
assert.Contains(t, full, "time")
|
|
assert.Contains(t, full, "source")
|
|
|
|
// Check the group was recorded for filtering
|
|
entries, _, _ = b.Get(0, math.MaxInt64, Filter{Level: slog.LevelDebug, Group: "job/1"}, 0)
|
|
decoded = decodeEntries(t, entries)
|
|
require.Len(t, decoded, 1)
|
|
assert.Equal(t, "hello", decoded[0].Msg)
|
|
}
|
|
|
|
func TestBufferGetGroup(t *testing.T) {
|
|
b := newTestBuffer(1024)
|
|
for i, jobID := range []int64{0, 1, 2, 1, 0, 2} {
|
|
attribution := Attribution{}
|
|
if jobID != 0 {
|
|
attribution = Attribution{JobID: jobID, Group: fmt.Sprintf("job/%d", jobID)}
|
|
}
|
|
add(b, slog.LevelInfo, attribution, fmt.Sprintf(`{"msg":"msg%d"}`, i))
|
|
}
|
|
add(b, slog.LevelDebug, Attribution{JobID: 1, Group: "job/1"}, `{"msg":"msg6"}`)
|
|
for _, test := range []struct {
|
|
name string
|
|
filter Filter
|
|
want []int64
|
|
}{
|
|
{"all", Filter{Level: slog.LevelDebug}, []int64{0, 1, 2, 3, 4, 5, 6}},
|
|
{"group", Filter{Level: slog.LevelDebug, Group: "job/1"}, []int64{1, 3, 6}},
|
|
{"group and unattributed", Filter{Level: slog.LevelDebug, Group: "job/1", Unattributed: true}, []int64{0, 1, 3, 4, 6}},
|
|
{"group and level", Filter{Level: slog.LevelInfo, Group: "job/1"}, []int64{1, 3}},
|
|
{"unknown group", Filter{Level: slog.LevelDebug, Group: "job/3"}, nil},
|
|
{"unknown group and unattributed", Filter{Level: slog.LevelDebug, Group: "job/3", Unattributed: true}, []int64{0, 4}},
|
|
{"job", Filter{Level: slog.LevelDebug, JobID: 1}, []int64{1, 3, 6}},
|
|
{"job and unattributed", Filter{Level: slog.LevelDebug, JobID: 2, Unattributed: true}, []int64{0, 2, 4, 5}},
|
|
{"job or group", Filter{Level: slog.LevelDebug, JobID: 2, Group: "job/1"}, []int64{1, 2, 3, 5, 6}},
|
|
} {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
entries, next, lost := b.Get(0, math.MaxInt64, test.filter, 0)
|
|
var got []int64
|
|
for _, e := range decodeEntries(t, entries) {
|
|
got = append(got, e.Seq)
|
|
}
|
|
assert.Equal(t, test.want, got)
|
|
assert.Equal(t, int64(7), next)
|
|
assert.Equal(t, int64(0), lost, "filtered entries aren't lost")
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestBufferConcurrency(t *testing.T) {
|
|
b := newTestBuffer(4096)
|
|
var wg sync.WaitGroup
|
|
for range 4 {
|
|
wg.Go(func() {
|
|
for i := range 500 {
|
|
addEntry(b, slog.LevelInfo, fmt.Sprintf("msg%d", i))
|
|
}
|
|
})
|
|
}
|
|
wg.Go(func() {
|
|
var from int64
|
|
for range 500 {
|
|
var entries []json.RawMessage
|
|
entries, from, _ = b.Get(from, math.MaxInt64, Filter{Level: slog.LevelDebug}, 10)
|
|
_ = decodeEntries(t, entries)
|
|
_ = b.Seq()
|
|
}
|
|
})
|
|
wg.Wait()
|
|
assert.Equal(t, int64(2000), b.Seq())
|
|
}
|
|
|
|
func TestRcLog(t *testing.T) {
|
|
call := rc.Calls.Get("core/log")
|
|
require.NotNil(t, call)
|
|
ctx := context.Background()
|
|
|
|
oldRecent := Recent
|
|
defer func() { Recent = oldRecent }()
|
|
Recent = newTestBuffer(0)
|
|
|
|
_, err := call.Fn(ctx, rc.Params{})
|
|
assert.Equal(t, ErrBufferDisabled, err)
|
|
assert.True(t, rc.IsErrParamInvalid(err), err)
|
|
|
|
Recent = newTestBuffer(64 * 1024)
|
|
for i := range 10 {
|
|
level := slog.LevelInfo
|
|
if i%2 == 1 {
|
|
level = slog.LevelError
|
|
}
|
|
addEntry(Recent, level, fmt.Sprintf("msg%d", i))
|
|
}
|
|
|
|
seqs := func(out rc.Params) (seqs []int64) {
|
|
for _, e := range decodeEntries(t, out["entries"].(Entries)) {
|
|
seqs = append(seqs, e.Seq)
|
|
}
|
|
return seqs
|
|
}
|
|
|
|
out, err := call.Fn(ctx, rc.Params{})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []int64{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, seqs(out))
|
|
assert.Equal(t, int64(10), out["next"])
|
|
assert.Equal(t, int64(0), out["lost"])
|
|
|
|
out, err = call.Fn(ctx, rc.Params{"since": 4, "limit": 2})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []int64{4, 5}, seqs(out))
|
|
assert.Equal(t, int64(6), out["next"])
|
|
|
|
out, err = call.Fn(ctx, rc.Params{"since": -3})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []int64{7, 8, 9}, seqs(out))
|
|
|
|
out, err = call.Fn(ctx, rc.Params{"since": -100, "level": "error"})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []int64{1, 3, 5, 7, 9}, seqs(out))
|
|
|
|
_, err = call.Fn(ctx, rc.Params{"level": "potato"})
|
|
require.Error(t, err)
|
|
assert.True(t, rc.IsErrParamInvalid(err), err)
|
|
|
|
// since at the end is OK but beyond the end is an error
|
|
out, err = call.Fn(ctx, rc.Params{"since": 10})
|
|
require.NoError(t, err)
|
|
assert.Empty(t, out["entries"])
|
|
assert.Equal(t, int64(10), out["next"])
|
|
_, err = call.Fn(ctx, rc.Params{"since": 11})
|
|
require.Error(t, err)
|
|
assert.True(t, rc.IsErrParamInvalid(err), err)
|
|
assert.Contains(t, err.Error(), "beyond the end")
|
|
|
|
// Check the output can be serialized
|
|
out, err = call.Fn(ctx, rc.Params{"since": 9})
|
|
require.NoError(t, err)
|
|
var buf []byte
|
|
buf, err = json.Marshal(out)
|
|
require.NoError(t, err)
|
|
assert.JSONEq(t, `{"entries":[{"seq":9,"level":"ERROR","msg":"msg9"}],"next":10,"lost":0}`, string(buf))
|
|
|
|
// Filter by group
|
|
addEntry(Recent, slog.LevelInfo, "not in a group")
|
|
add(Recent, slog.LevelInfo, Attribution{JobID: 1, Group: "job/1"}, `{"msg":"in job/1"}`)
|
|
out, err = call.Fn(ctx, rc.Params{"since": 10, "group": "job/1"})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []int64{11}, seqs(out))
|
|
assert.Equal(t, int64(12), out["next"])
|
|
|
|
// Filter by rc job
|
|
out, err = call.Fn(ctx, rc.Params{"since": 10, "jobid": 1})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, []int64{11}, seqs(out))
|
|
}
|