1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package cilogs_test
4
5import (
6 "context"
7 "os"
8 "path/filepath"
9 "strings"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/assert"
14 "github.com/stretchr/testify/require"
15
16 "bnema.dev/ermite/internal/adapters/cilogs"
17 "bnema.dev/ermite/internal/domain"
18 "bnema.dev/ermite/internal/ports"
19)
20
21var (
22 ref0 = ports.CILogRef{RunID: 7, JobID: "7-0-0", Attempt: 1}
23 t0 = time.Date(2026, 10, 12, 10, 0, 0, 123456789, time.UTC)
24)
25
26func lines(texts ...string) []domain.CILogLine {
27 out := make([]domain.CILogLine, len(texts))
28 for i, s := range texts {
29 out[i] = domain.CILogLine{Time: t0.Add(time.Duration(i) * time.Millisecond), Text: s}
30 }
31 return out
32}
33
34const big = 1 << 30
35
36func TestAppendAndReadKeepTextTimeAndContinuation(t *testing.T) {
37 ctx := context.Background()
38 s := cilogs.New(t.TempDir())
39 in := lines("##[group]Run echo hi", "hi", "##[endgroup]")
40 in = append(in, domain.CILogLine{Time: t0, Text: "second half", Continued: true})
41 next, added, err := s.Append(ctx, ref0, 0, in, big)
42 require.NoError(t, err)
43 assert.Equal(t, int64(4), next)
44 assert.Equal(t, 4, added)
45
46 got, err := s.Read(ctx, ref0, 0, 100, 1<<20)
47 require.NoError(t, err)
48 require.Len(t, got, 4)
49 assert.Equal(t, "##[group]Run echo hi", got[0].Text)
50 assert.True(t, got[0].Time.Equal(t0), "nanosecond precision is kept: %v", got[0].Time)
51 assert.True(t, got[3].Continued)
52 assert.False(t, got[2].Continued)
53
54 tail, err := s.Read(ctx, ref0, 2, 100, 1<<20)
55 require.NoError(t, err)
56 assert.Equal(t, "##[endgroup]", tail[0].Text)
57}
58
59func TestAppendIsIdempotentByOffset(t *testing.T) {
60 ctx := context.Background()
61 s := cilogs.New(t.TempDir())
62 _, _, err := s.Append(ctx, ref0, 0, lines("a", "b", "c"), big)
63 require.NoError(t, err)
64
65 // A retry of the same request changes nothing.
66 next, added, err := s.Append(ctx, ref0, 0, lines("a", "b", "c"), big)
67 require.NoError(t, err)
68 assert.Equal(t, int64(3), next)
69 assert.Zero(t, added)
70
71 // A request that overlaps appends only its new tail.
72 next, added, err = s.Append(ctx, ref0, 1, lines("b", "c", "d"), big)
73 require.NoError(t, err)
74 assert.Equal(t, int64(4), next)
75 assert.Equal(t, 1, added)
76
77 // A retry whose lines differ is silently dropped, as the contract says.
78 next, added, err = s.Append(ctx, ref0, 0, lines("X", "Y"), big)
79 require.NoError(t, err)
80 assert.Equal(t, int64(4), next)
81 assert.Zero(t, added)
82 got, _ := s.Read(ctx, ref0, 0, 10, 1<<20)
83 assert.Equal(t, "a", got[0].Text)
84}
85
86func TestAppendAboveTheCountIsAGap(t *testing.T) {
87 ctx := context.Background()
88 s := cilogs.New(t.TempDir())
89 _, _, err := s.Append(ctx, ref0, 0, lines("a"), big)
90 require.NoError(t, err)
91 _, _, err = s.Append(ctx, ref0, 5, lines("z"), big)
92 require.ErrorIs(t, err, domain.ErrOutOfRange)
93 var gap *ports.ErrCILogGap
94 require.ErrorAs(t, err, &gap)
95 assert.Equal(t, int64(1), gap.Count)
96}
97
98func TestTheLimitRefusesTheWholeRequestUnlessForced(t *testing.T) {
99 ctx := context.Background()
100 s := cilogs.New(t.TempDir())
101 _, _, err := s.Append(ctx, ref0, 0, lines("aaaa"), 200)
102 require.NoError(t, err)
103 _, _, err = s.Append(ctx, ref0, 1, lines(strings.Repeat("b", 300)), 200)
104 require.ErrorIs(t, err, ports.ErrCILogLimit)
105 n, _ := s.Count(ctx, ref0)
106 assert.Equal(t, int64(1), n, "nothing of the refused request was appended")
107
108 next, err := s.AppendForced(ctx, ref0, lines("limit reached"))
109 require.NoError(t, err)
110 assert.Equal(t, int64(2), next)
111}
112
113func TestReadBoundsLinesAndBytesAndReportsMissingLogs(t *testing.T) {
114 ctx := context.Background()
115 s := cilogs.New(t.TempDir())
116 _, err := s.Read(ctx, ref0, 0, 10, 100)
117 require.ErrorIs(t, err, domain.ErrNotFound)
118
119 _, _, err = s.Append(ctx, ref0, 0, lines("12345", "12345", "12345", "12345"), big)
120 require.NoError(t, err)
121 got, err := s.Read(ctx, ref0, 0, 3, 1<<20)
122 require.NoError(t, err)
123 assert.Len(t, got, 3)
124 got, err = s.Read(ctx, ref0, 0, 100, 12)
125 require.NoError(t, err)
126 assert.Len(t, got, 2, "the text of a batch is bounded")
127 got, err = s.Read(ctx, ref0, 4, 10, 100)
128 require.NoError(t, err)
129 assert.Empty(t, got, "reading at the end is empty, not an error")
130 _, err = s.Read(ctx, ref0, 5, 10, 100)
131 assert.ErrorIs(t, err, domain.ErrOutOfRange)
132}
133
134func TestReadFindsLinesDeepInALargeLog(t *testing.T) {
135 ctx := context.Background()
136 s := cilogs.New(t.TempDir())
137 var all []domain.CILogLine
138 for i := range 3000 {
139 all = append(all, domain.CILogLine{Time: t0, Text: "line " + itoa(i)})
140 }
141 for off := 0; off < len(all); off += 1000 {
142 _, _, err := s.Append(ctx, ref0, int64(off), all[off:off+1000], big)
143 require.NoError(t, err)
144 }
145 got, err := s.Read(ctx, ref0, 2500, 2, 1<<20)
146 require.NoError(t, err)
147 require.Len(t, got, 2)
148 assert.Equal(t, "line 2500", got[0].Text)
149 assert.Equal(t, "line 2501", got[1].Text)
150}
151
152func itoa(n int) string {
153 if n == 0 {
154 return "0"
155 }
156 var b []byte
157 for n > 0 {
158 b = append([]byte{byte('0' + n%10)}, b...)
159 n /= 10
160 }
161 return string(b)
162}
163
164func TestCountSurvivesARestartAndAChangedChannelFiresOnAppend(t *testing.T) {
165 ctx := context.Background()
166 dir := t.TempDir()
167 s := cilogs.New(dir)
168 ch := s.Changed(ref0)
169 _, _, err := s.Append(ctx, ref0, 0, lines("a", "b"), big)
170 require.NoError(t, err)
171 select {
172 case <-ch:
173 default:
174 t.Fatal("Changed did not fire on append")
175 }
176
177 s2 := cilogs.New(dir)
178 n, err := s2.Count(ctx, ref0)
179 require.NoError(t, err)
180 assert.Equal(t, int64(2), n)
181 _, _, err = s2.Append(ctx, ref0, 2, lines("c"), big)
182 require.NoError(t, err)
183 got, _ := s2.Read(ctx, ref0, 0, 10, 1<<20)
184 assert.Len(t, got, 3)
185}
186
187func TestATornLastLineIsCutOnLoad(t *testing.T) {
188 ctx := context.Background()
189 dir := t.TempDir()
190 s := cilogs.New(dir)
191 _, _, err := s.Append(ctx, ref0, 0, lines("a", "b"), big)
192 require.NoError(t, err)
193 path := filepath.Join(dir, "7", "7-0-0", "1.log")
194 f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0)
195 require.NoError(t, err)
196 _, err = f.WriteString("- 2026-10-12T10:00:00.1234")
197 require.NoError(t, err)
198 require.NoError(t, f.Close())
199
200 s2 := cilogs.New(dir)
201 n, err := s2.Count(ctx, ref0)
202 require.NoError(t, err)
203 assert.Equal(t, int64(2), n)
204 _, _, err = s2.Append(ctx, ref0, 2, lines("c"), big)
205 require.NoError(t, err)
206 got, err := s2.Read(ctx, ref0, 0, 10, 1<<20)
207 require.NoError(t, err)
208 assert.Len(t, got, 3)
209}
210
211func TestRemoveRunsAndResetAndRunIDs(t *testing.T) {
212 ctx := context.Background()
213 s := cilogs.New(t.TempDir())
214 other := ports.CILogRef{RunID: 9, JobID: "9-0-0", Attempt: 1}
215 _, _, _ = s.Append(ctx, ref0, 0, lines("a"), big)
216 _, _, _ = s.Append(ctx, other, 0, lines("b"), big)
217 ids, err := s.RunIDs(ctx)
218 require.NoError(t, err)
219 assert.ElementsMatch(t, []int64{7, 9}, ids)
220
221 require.NoError(t, s.Reset(ctx, ref0))
222 n, _ := s.Count(ctx, ref0)
223 assert.Zero(t, n)
224
225 require.NoError(t, s.RemoveRuns(ctx, []int64{9}))
226 _, err = s.Read(ctx, other, 0, 10, 100)
227 assert.ErrorIs(t, err, domain.ErrNotFound)
228}
229
230func TestARefThatCouldEscapeTheDirectoryIsRefused(t *testing.T) {
231 ctx := context.Background()
232 dir := t.TempDir()
233 s := cilogs.New(dir)
234 _, _, err := s.Append(ctx, ports.CILogRef{RunID: 1, JobID: "../../etc", Attempt: 1}, 0, lines("x"), big)
235 assert.ErrorIs(t, err, domain.ErrNotFound)
236 _, _, err = s.Append(ctx, ports.CILogRef{RunID: 1, JobID: "1-0-0", Attempt: 0}, 0, lines("x"), big)
237 assert.ErrorIs(t, err, domain.ErrNotFound)
238}