1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
|
package main
import (
"database/sql"
"errors"
"fmt"
"net/http"
"net/url"
"strings"
"testing"
"time"
)
// makeIssue builds an issue fixture. `pull` makes it a pull request, which in
// Forgejo means the same object with pull_request populated.
func makeIssue(number int64, title, state, updated string, pull bool, labels ...apiLabel) apiIssue {
i := apiIssue{
ID: 1000 + number,
Number: number,
Title: title,
Body: "body of " + title,
State: state,
Labels: labels,
CreatedAt: ts(1, 1, 12, "+01:00"),
UpdatedAt: updated,
}
i.User.Login = "alice"
if pull {
i.PullRequest = &struct {
Merged bool `json:"merged"`
Draft bool `json:"draft"`
}{}
}
return i
}
func makeComment(id, number int64, body, updated string, onPull bool) apiComment {
c := apiComment{
ID: id,
Body: body,
CreatedAt: updated,
UpdatedAt: updated,
HTMLURL: fmt.Sprintf("https://fake.test/acme/widget/issues/%d#issuecomment-%d", number, id),
}
c.User.Login = "bob"
// Exactly one of the two is set, and for a PR comment it is the pull
// URL while issue_url is empty — this is the shape commentNumber has to
// cope with.
if onPull {
c.PullRequestURL = fmt.Sprintf("https://fake.test/api/v1/repos/acme/widget/pulls/%d", number)
} else {
c.IssueURL = fmt.Sprintf("https://fake.test/api/v1/repos/acme/widget/issues/%d", number)
}
return c
}
// A full ingest must land every kind of row in the right table, and in
// particular must not put pull requests somewhere other than `issue`.
func TestIngestFull(t *testing.T) {
f := newFakeForge(t)
bug := apiLabel{ID: 7, Name: "bug", Color: "ff0000", Description: "a defect"}
f.labels = []apiLabel{bug}
f.issues = []apiIssue{
makeIssue(1, "an issue", "open", ts(3, 1, 10, "+01:00"), false, bug),
makeIssue(2, "a pull request", "closed", ts(3, 2, 10, "+01:00"), true),
makeIssue(3, "another issue", "open", ts(3, 3, 10, "+01:00"), false),
}
merged := ts(3, 2, 11, "+01:00")
f.pulls = []apiPull{{
Number: 2, Merged: true, MergedAt: &merged,
MergedBy: &dLogin{Login: "carol"},
Additions: ptr(int64(28)), Deletions: ptr(int64(1)), ChangedFiles: ptr(int64(2)),
UpdatedAt: ts(3, 2, 10, "+01:00"),
}}
f.pulls[0].Head.Ref = "feature"
f.pulls[0].Base.Ref = "master"
f.comments = []apiComment{
makeComment(500, 1, "on the issue", ts(3, 1, 11, "+01:00"), false),
makeComment(501, 2, "on the PR", ts(3, 2, 12, "+01:00"), true),
}
f.reviews[2] = []apiReview{{
ID: 900, State: "APPROVED", CommentsCount: 1,
SubmittedAt: ts(3, 2, 10, "+01:00"),
User: &dLogin{Login: "carol"},
}}
f.inline[900] = []apiReviewComment{{
ID: 950, ReviewID: 900, Body: "this line is wrong",
Path: "src/main.go", Position: ptr(int64(12)),
DiffHunk: "@@ -1 +1 @@", CommitID: "abc123",
CreatedAt: ts(3, 2, 10, "+01:00"), UpdatedAt: ts(3, 2, 10, "+01:00"),
User: &dLogin{Login: "carol"},
Resolver: &dLogin{Login: "alice"},
}}
db := testDB(t)
ing := f.ingester(t, db)
runAllPhases(t, ing)
if got := mustCount(t, db, `SELECT count(*) FROM issue WHERE is_pull = 0`); got != 2 {
t.Errorf("issues = %d, want 2", got)
}
if got := mustCount(t, db, `SELECT count(*) FROM issue WHERE is_pull = 1`); got != 1 {
t.Errorf("pull requests in issue table = %d, want 1", got)
}
if got := mustCount(t, db, `SELECT count(*) FROM pull`); got != 1 {
t.Errorf("pull rows = %d, want 1", got)
}
if got := mustCount(t, db, `SELECT count(*) FROM comment`); got != 2 {
t.Errorf("comments = %d, want 2", got)
}
if got := mustCount(t, db, `SELECT count(*) FROM review`); got != 1 {
t.Errorf("reviews = %d, want 1", got)
}
if got := mustCount(t, db, `SELECT count(*) FROM review_comment`); got != 1 {
t.Errorf("review comments = %d, want 1", got)
}
// A comment on a PR must be attached to the PR's number, which is only
// recoverable from pull_request_url.
if got := mustCount(t, db, `SELECT number FROM comment WHERE id = 501`); got != 2 {
t.Errorf("PR comment attached to issue %d, want 2", got)
}
// The label join, not a JSON blob.
if got := mustCount(t, db,
`SELECT count(*) FROM issue_label il JOIN label l ON l.id = il.label_id
WHERE l.name = 'bug'`); got != 1 {
t.Errorf("issues labelled bug = %d, want 1", got)
}
// Diff stats and the resolver must survive.
var add, del int64
var mergedBy string
if err := db.QueryRow(`SELECT additions, deletions, merged_by FROM pull WHERE number = 2`).
Scan(&add, &del, &mergedBy); err != nil {
t.Fatal(err)
}
if add != 28 || del != 1 || mergedBy != "carol" {
t.Errorf("pull 2: +%d -%d by %q, want +28 -1 by carol", add, del, mergedBy)
}
var path, resolver string
if err := db.QueryRow(`SELECT path, resolver FROM review_comment WHERE id = 950`).
Scan(&path, &resolver); err != nil {
t.Fatal(err)
}
if path != "src/main.go" || resolver != "alice" {
t.Errorf("review comment: path=%q resolver=%q", path, resolver)
}
}
// Running the same ingest twice must not duplicate anything. This is what
// lets the tool be run from a shell loop without thinking about it.
func TestIngestIsIdempotent(t *testing.T) {
f := newFakeForge(t)
f.issues = []apiIssue{
makeIssue(1, "an issue", "open", ts(3, 1, 10, "+01:00"), false),
makeIssue(2, "a pull request", "open", ts(3, 2, 10, "+01:00"), true),
}
f.pulls = []apiPull{{Number: 2, UpdatedAt: ts(3, 2, 10, "+01:00")}}
f.comments = []apiComment{makeComment(500, 1, "hi", ts(3, 1, 11, "+01:00"), false)}
f.reviews[2] = []apiReview{{ID: 900, State: "APPROVED", SubmittedAt: ts(3, 2, 10, "+01:00")}}
db := testDB(t)
ing := f.ingester(t, db)
runAllPhases(t, ing)
before := snapshot(t, db)
// A second ingester, as a fresh process would be.
runAllPhases(t, f.ingester(t, db))
after := snapshot(t, db)
if before != after {
t.Errorf("second ingest changed the database:\n before %v\n after %v", before, after)
}
}
// The incremental path: what changed is re-fetched and lands, and what did
// not is left alone.
//
// The boundary is deliberately inclusive. `since` is passed as the greatest
// updated_at we have seen, and the API's filter is >=, so the item sitting
// exactly on the watermark comes back every run. That is the intended trade:
// an exclusive filter would need us to add a second to the mark, which would
// silently skip anything modified within that second. One redundant UPSERT
// per run is the cheaper mistake.
func TestIngestIncremental(t *testing.T) {
f := newFakeForge(t)
f.issues = []apiIssue{
makeIssue(1, "old issue", "open", ts(3, 1, 10, "+01:00"), false),
makeIssue(2, "untouched issue", "open", ts(3, 2, 10, "+01:00"), false),
makeIssue(3, "the newest thing", "open", ts(3, 3, 10, "+01:00"), true),
}
f.pulls = []apiPull{{Number: 3, UpdatedAt: ts(3, 3, 10, "+01:00")}}
db := testDB(t)
runAllPhases(t, f.ingester(t, db))
// The watermark is now issue 3's timestamp, the greatest seen.
// Upstream moves on: issue 1 is edited and closed, issue 4 appears.
f.issues[0].Title = "old issue, retitled"
f.issues[0].State = "closed"
f.issues[0].UpdatedAt = ts(4, 1, 10, "+02:00") // note: summer offset
f.issues = append(f.issues, makeIssue(4, "new issue", "open", ts(4, 2, 10, "+02:00"), false))
ing := f.ingester(t, db)
runAllPhases(t, ing)
// Issues 1 and 4 changed; issue 3 sits exactly on the watermark and is
// re-fetched by the inclusive filter. Issue 2 is strictly older and must
// NOT have been touched — that is what makes a refresh cheap.
if ing.nIssues != 3 {
t.Errorf("second run wrote %d issues, want 3 (two changed + the one on the watermark)", ing.nIssues)
}
var title, state string
if err := db.QueryRow(`SELECT title, state FROM issue WHERE number = 1`).Scan(&title, &state); err != nil {
t.Fatal(err)
}
if title != "old issue, retitled" || state != "closed" {
t.Errorf("issue 1 = %q/%q, want the updated title and closed", title, state)
}
if got := mustCount(t, db, `SELECT count(*) FROM issue`); got != 4 {
t.Errorf("issues = %d, want 4", got)
}
// A third run with nothing changed upstream must write only the
// watermark item, not walk the whole repo again.
third := f.ingester(t, db)
runAllPhases(t, third)
if third.nIssues > 1 {
t.Errorf("third run on an unchanged repo wrote %d issues, want at most 1", third.nIssues)
}
}
// A watermark must be derived from the data, not the clock, and must survive
// the offset change between winter and summer time. Codeberg returns +01:00
// in January and +02:00 in July, so "2026-07-01T00:00:00+02:00" is LATER than
// "2026-06-30T23:30:00+01:00" but sorts earlier as a string.
func TestWatermarkIsChronologicalNotLexical(t *testing.T) {
winter := "2026-01-08T20:00:00+01:00"
summer := "2026-07-08T20:00:00+02:00"
if !later(summer, winter) {
t.Errorf("later(%q, %q) = false, want true", summer, winter)
}
// The pathological pair: string comparison gets this backwards.
early := "2026-06-30T23:30:00+01:00" // 22:30 UTC
late := "2026-07-01T00:00:00+02:00" // 22:00 UTC — actually EARLIER
if later(late, early) {
t.Errorf("later(%q, %q) = true, but %s is the earlier instant", late, early, late)
}
if mustParse(t, late).After(mustParse(t, early)) {
t.Fatal("fixture is wrong; check the offsets")
}
// laterOf must agree.
if got := laterOf(winter, summer); got != summer {
t.Errorf("laterOf = %q, want %q", got, summer)
}
if got := laterOf("", winter); got != winter {
t.Errorf("laterOf with empty = %q, want %q", got, winter)
}
}
// Pagination is the part most likely to break silently, so it is tested for
// the exact off-by-one that a short-page terminator has to get right: a
// collection that is an exact multiple of the page size.
func TestPaginationExactMultiple(t *testing.T) {
f := newFakeForge(t)
for n := int64(1); n <= int64(2*pageSize); n++ {
f.issues = append(f.issues, makeIssue(n, fmt.Sprintf("issue %d", n), "open",
ts(3, 1, 10, "+01:00"), false))
}
db := testDB(t)
ing := f.ingester(t, db)
if err := ing.repo(); err != nil {
t.Fatal(err)
}
if err := ing.issues(); err != nil {
t.Fatal(err)
}
if got := mustCount(t, db, `SELECT count(*) FROM issue`); got != int64(2*pageSize) {
t.Errorf("issues = %d, want %d", got, 2*pageSize)
}
}
// getList must ALWAYS send ?page. Forgejo only paginates when it is present:
// without it the server returns the whole collection and ignores ?limit,
// which works by accident until the collection is big enough to be truncated
// upstream. The fake reproduces that, so this test fails loudly if the
// parameter is ever dropped.
func TestGetListAlwaysPaginates(t *testing.T) {
f := newFakeForge(t)
for n := int64(1); n <= 120; n++ {
f.issues = append(f.issues, makeIssue(n, "x", "open", ts(3, 1, 10, "+01:00"), false))
}
var seen []string
f.srv.Config.Handler = wrapHandler(f.srv.Config.Handler, func(u *url.URL) {
if strings.HasSuffix(u.Path, "/issues") {
seen = append(seen, u.Query().Get("page"))
}
})
c := f.client()
n := 0
err := getList(c, "/repos/acme/widget/issues", url.Values{}, func(b []apiIssue) error {
n += len(b)
return nil
})
if err != nil {
t.Fatal(err)
}
if n != 120 {
t.Errorf("fetched %d issues, want 120", n)
}
for i, p := range seen {
if p == "" {
t.Fatalf("request %d omitted ?page; Forgejo would have returned the whole collection", i+1)
}
}
if len(seen) != 3 { // 50 + 50 + 20
t.Errorf("made %d requests (%v), want 3", len(seen), seen)
}
}
// The reviews phase must only ask about numbers that are pull requests.
// /pulls/{n}/reviews on an issue is an error, and a run that hits it for
// every issue in the repo both wastes the rate limit and fails.
func TestReviewsSkipsIssues(t *testing.T) {
f := newFakeForge(t)
f.issues = []apiIssue{
makeIssue(1, "an issue", "open", ts(3, 1, 10, "+01:00"), false),
makeIssue(2, "a pull", "open", ts(3, 2, 10, "+01:00"), true),
}
f.pulls = []apiPull{{Number: 2, UpdatedAt: ts(3, 2, 10, "+01:00")}}
f.reviews[2] = []apiReview{{ID: 900, State: "APPROVED", SubmittedAt: ts(3, 2, 10, "+01:00")}}
db := testDB(t)
ing := f.ingester(t, db)
runAllPhases(t, ing)
if n := f.requests["/pulls/1/reviews"]; n != 0 {
t.Errorf("asked for the reviews of issue 1 %d times; that endpoint errors on issues", n)
}
if n := f.requests["/pulls/2/reviews"]; n != 1 {
t.Errorf("asked for the reviews of pull 2 %d times, want 1", n)
}
}
// A review deleted upstream must disappear locally. The phase replaces a PR's
// reviews rather than upserting, precisely so that this works; an
// upsert-only implementation would keep the ghost for ever.
func TestReviewsAreReplacedNotAccumulated(t *testing.T) {
f := newFakeForge(t)
f.issues = []apiIssue{makeIssue(2, "a pull", "open", ts(3, 2, 10, "+01:00"), true)}
f.pulls = []apiPull{{Number: 2, UpdatedAt: ts(3, 2, 10, "+01:00")}}
f.reviews[2] = []apiReview{
{ID: 900, State: "APPROVED", SubmittedAt: ts(3, 2, 10, "+01:00")},
{ID: 901, State: "COMMENT", CommentsCount: 1, SubmittedAt: ts(3, 2, 11, "+01:00")},
}
f.inline[901] = []apiReviewComment{{
ID: 950, ReviewID: 901, Path: "a.go", Body: "hm",
CreatedAt: ts(3, 2, 11, "+01:00"), UpdatedAt: ts(3, 2, 11, "+01:00"),
}}
db := testDB(t)
runAllPhases(t, f.ingester(t, db))
if got := mustCount(t, db, `SELECT count(*) FROM review`); got != 2 {
t.Fatalf("reviews = %d, want 2", got)
}
// The second review is dismissed upstream, and the PR is touched.
f.reviews[2] = f.reviews[2][:1]
delete(f.inline, 901)
f.issues[0].UpdatedAt = ts(3, 5, 10, "+01:00")
f.pulls[0].UpdatedAt = ts(3, 5, 10, "+01:00")
runAllPhases(t, f.ingester(t, db))
if got := mustCount(t, db, `SELECT count(*) FROM review`); got != 1 {
t.Errorf("reviews = %d after one was removed upstream, want 1", got)
}
if got := mustCount(t, db, `SELECT count(*) FROM review_comment`); got != 0 {
t.Errorf("review comments = %d, want 0", got)
}
}
// An interrupted review backfill must resume rather than restart: it is the
// one phase that costs a request per pull request.
func TestReviewBackfillResumes(t *testing.T) {
f := newFakeForge(t)
for n := int64(1); n <= 5; n++ {
f.issues = append(f.issues, makeIssue(n, "pull", "open", ts(3, int(n), 10, "+01:00"), true))
f.pulls = append(f.pulls, apiPull{Number: n, UpdatedAt: ts(3, int(n), 10, "+01:00")})
f.reviews[n] = []apiReview{{ID: 900 + n, State: "APPROVED", SubmittedAt: ts(3, int(n), 10, "+01:00")}}
}
db := testDB(t)
ing := f.ingester(t, db)
if err := ing.repo(); err != nil {
t.Fatal(err)
}
if err := ing.issues(); err != nil {
t.Fatal(err)
}
// Simulate a backfill that got as far as pull 3 and was killed.
for _, n := range []int64{1, 2, 3} {
if _, err := ing.reviewsOf(n); err != nil {
t.Fatal(err)
}
}
if err := setMeta(db, "reviews:through", "3"); err != nil {
t.Fatal(err)
}
resumed := f.ingester(t, db)
if err := resumed.repo(); err != nil {
t.Fatal(err)
}
for k := range f.requests {
delete(f.requests, k)
}
if err := resumed.reviews(); err != nil {
t.Fatal(err)
}
for _, n := range []int64{1, 2, 3} {
if got := f.requests[fmt.Sprintf("/pulls/%d/reviews", n)]; got != 0 {
t.Errorf("re-fetched pull %d, which the backfill had already done", n)
}
}
for _, n := range []int64{4, 5} {
if got := f.requests[fmt.Sprintf("/pulls/%d/reviews", n)]; got != 1 {
t.Errorf("pull %d fetched %d times, want 1", n, got)
}
}
if got := mustCount(t, db, `SELECT count(*) FROM review`); got != 5 {
t.Errorf("reviews = %d, want 5", got)
}
// The resume point must be cleared once the backfill completes, or a
// later --full run would skip everything below it.
if v, _ := getMeta(db, "reviews:through"); v != "" {
t.Errorf("reviews:through = %q after completion, want empty", v)
}
}
// /pulls ignores ?since, so the phase must stop by itself once it has walked
// past the watermark instead of paging through every PR in the repo.
func TestPullsStopEarlyOnIncrementalRun(t *testing.T) {
f := newFakeForge(t)
for n := int64(1); n <= 120; n++ {
u := ts(3, 1, 10, "+01:00")
f.issues = append(f.issues, makeIssue(n, "pull", "open", u, true))
f.pulls = append(f.pulls, apiPull{Number: n, UpdatedAt: u})
}
db := testDB(t)
ing := f.ingester(t, db)
if err := ing.repo(); err != nil {
t.Fatal(err)
}
if err := ing.issues(); err != nil {
t.Fatal(err)
}
if err := ing.pulls(); err != nil {
t.Fatal(err)
}
// Nothing changed upstream; the next run should stop after one page.
f.requests["/pulls"] = 0
second := f.ingester(t, db)
if err := second.repo(); err != nil {
t.Fatal(err)
}
if err := second.pulls(); err != nil {
t.Fatal(err)
}
if got := f.requests["/pulls"]; got != 1 {
t.Errorf("made %d /pulls requests on an unchanged repo, want 1 (early exit)", got)
}
if second.nPulls != 0 {
t.Errorf("wrote %d pull rows on an unchanged repo, want 0", second.nPulls)
}
}
// The comment feed has its own watermark and can name an issue the issue
// phase has not reached. Such a comment must be KEPT — the next run's `since`
// has already moved past it, so dropping it loses it for ever — and reported.
func TestCommentOnUnknownIssueIsKept(t *testing.T) {
f := newFakeForge(t)
f.issues = []apiIssue{makeIssue(1, "an issue", "open", ts(3, 1, 10, "+01:00"), false)}
f.comments = []apiComment{
makeComment(500, 1, "fine", ts(3, 1, 11, "+01:00"), false),
// Issue 99 is not in the issue list: it was created after that phase
// ran.
makeComment(501, 99, "on an issue we have not seen", ts(3, 1, 12, "+01:00"), false),
}
db := testDB(t)
ing := f.ingester(t, db)
runAllPhases(t, ing)
if got := mustCount(t, db, `SELECT count(*) FROM comment`); got != 2 {
t.Errorf("comments = %d, want 2 — the orphan must be kept, not dropped", got)
}
if got := mustCount(t, db, `SELECT number FROM comment WHERE id = 501`); got != 99 {
t.Errorf("orphan comment attached to %d, want 99", got)
}
}
// commentNumber is the only way to know what a comment belongs to: the object
// carries no number field, just two URLs of which one is empty.
func TestCommentNumber(t *testing.T) {
cases := []struct {
name string
c apiComment
want int64
wantOK bool
}{
{"issue", apiComment{IssueURL: "https://codeberg.org/api/v1/repos/a/b/issues/1065"}, 1065, true},
{"pull", apiComment{PullRequestURL: "https://codeberg.org/api/v1/repos/a/b/pulls/925"}, 925, true},
{"both, issue wins", apiComment{
IssueURL: "https://codeberg.org/api/v1/repos/a/b/issues/1",
PullRequestURL: "https://codeberg.org/api/v1/repos/a/b/pulls/2",
}, 1, true},
{"neither", apiComment{}, 0, false},
{"junk", apiComment{IssueURL: "https://codeberg.org/api/v1/repos/a/b/issues/abc"}, 0, false},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got, ok := commentNumber(tc.c)
if got != tc.want || ok != tc.wantOK {
t.Errorf("= %d, %v; want %d, %v", got, ok, tc.want, tc.wantOK)
}
})
}
}
// A label removed from an issue upstream must be removed here too. There is
// no "label removed" signal, so the phase replaces an issue's labels; this
// checks it actually does.
func TestLabelsAreReplaced(t *testing.T) {
f := newFakeForge(t)
bug := apiLabel{ID: 7, Name: "bug", Color: "ff0000"}
docs := apiLabel{ID: 8, Name: "docs", Color: "00ff00"}
f.labels = []apiLabel{bug, docs}
f.issues = []apiIssue{makeIssue(1, "x", "open", ts(3, 1, 10, "+01:00"), false, bug, docs)}
db := testDB(t)
runAllPhases(t, f.ingester(t, db))
if got := mustCount(t, db, `SELECT count(*) FROM issue_label WHERE number = 1`); got != 2 {
t.Fatalf("labels = %d, want 2", got)
}
f.issues[0].Labels = []apiLabel{docs}
f.issues[0].UpdatedAt = ts(3, 2, 10, "+01:00")
runAllPhases(t, f.ingester(t, db))
if got := mustCount(t, db, `SELECT count(*) FROM issue_label WHERE number = 1`); got != 1 {
t.Errorf("labels = %d after one was removed upstream, want 1", got)
}
if got := mustCount(t, db,
`SELECT count(*) FROM issue_label il JOIN label l ON l.id=il.label_id
WHERE il.number=1 AND l.name='docs'`); got != 1 {
t.Error("the surviving label should be docs")
}
}
// An error object where a list was expected must surface Forgejo's message,
// not "cannot unmarshal object into Go value of type []main.apiReview".
func TestAPIErrorMessageIsSurfaced(t *testing.T) {
f := newFakeForge(t)
f.issues = []apiIssue{makeIssue(1, "an issue", "open", ts(3, 1, 10, "+01:00"), false)}
db := testDB(t)
ing := f.ingester(t, db)
if err := ing.repo(); err != nil {
t.Fatal(err)
}
// Ask for the reviews of an issue, which is what the real API rejects.
err := getList(ing.c, ing.api("/pulls/1/reviews"), nil, func([]apiReview) error { return nil })
if err == nil {
t.Fatal("expected an error")
}
if !strings.Contains(err.Error(), "GetPullRequestByIndex") {
t.Errorf("error = %v, want it to carry the API's own message", err)
}
}
// A repository can have pull requests switched off — Codeberg/Community is
// one — and Forgejo then answers 404 on /pulls rather than an empty list.
// That is a capability, not an error, and must not abort a run that can still
// mirror the issues perfectly well.
func TestRepoWithPullRequestsDisabled(t *testing.T) {
f := newFakeForge(t)
f.repo.HasPullRequests = false
f.issues = []apiIssue{
makeIssue(1, "an issue", "open", ts(3, 1, 10, "+01:00"), false),
makeIssue(2, "another", "closed", ts(3, 2, 10, "+01:00"), false),
}
f.comments = []apiComment{makeComment(500, 1, "hello", ts(3, 1, 11, "+01:00"), false)}
db := testDB(t)
ing := f.ingester(t, db)
runAllPhases(t, ing) // must not error
if got := mustCount(t, db, `SELECT count(*) FROM issue`); got != 2 {
t.Errorf("issues = %d, want 2 — issues must still be mirrored", got)
}
if got := mustCount(t, db, `SELECT count(*) FROM comment`); got != 1 {
t.Errorf("comments = %d, want 1", got)
}
if n := f.requests["/pulls"]; n != 0 {
t.Errorf("requested /pulls %d times on a repo with pull requests disabled", n)
}
}
// Codeberg meters /pulls/{n}/reviews in a "git_op" bucket of 250 requests per
// ten minutes, separate from the 2000 that everything else gets. A cold
// backfill of a repo with more pull requests than that MUST survive hitting
// the wall, because it is guaranteed to.
func TestReviewBackfillWaitsOutRateLimit(t *testing.T) {
f := newFakeForge(t)
for n := int64(1); n <= 4; n++ {
f.issues = append(f.issues, makeIssue(n, "pull", "open", ts(3, int(n), 10, "+01:00"), true))
f.pulls = append(f.pulls, apiPull{Number: n, UpdatedAt: ts(3, int(n), 10, "+01:00")})
f.reviews[n] = []apiReview{{ID: 900 + n, State: "APPROVED", SubmittedAt: ts(3, int(n), 10, "+01:00")}}
}
// Refuse the third review request once, as the real API does mid-backfill.
f.rateLimitReviewsOnce = 3
db := testDB(t)
ing := f.ingester(t, db)
var slept []time.Duration
ing.c.sleep = func(d time.Duration) { slept = append(slept, d) }
runAllPhases(t, ing)
if len(slept) != 1 {
t.Fatalf("slept %d times, want 1", len(slept))
}
// The wait must come from the header's t=, not a hardcoded guess.
if slept[0] < 600*time.Second {
t.Errorf("slept %s, want at least the advertised 600s window", slept[0])
}
if got := mustCount(t, db, `SELECT count(*) FROM review`); got != 4 {
t.Errorf("reviews = %d, want 4 — the backfill must complete after waiting", got)
}
}
// An incremental run must NOT wait: it has a handful of PRs to check, so a
// rate limit means something is wrong, and blocking a terminal for ten
// minutes in silence is worse than an error that says what happened.
func TestIncrementalReviewsDoNotWait(t *testing.T) {
f := newFakeForge(t)
f.issues = []apiIssue{makeIssue(1, "pull", "open", ts(3, 1, 10, "+01:00"), true)}
f.pulls = []apiPull{{Number: 1, UpdatedAt: ts(3, 1, 10, "+01:00")}}
f.reviews[1] = []apiReview{{ID: 901, State: "APPROVED", SubmittedAt: ts(3, 1, 10, "+01:00")}}
db := testDB(t)
runAllPhases(t, f.ingester(t, db)) // cold run establishes the watermark
// Touch the PR so the incremental run has something to do, then make the
// API refuse it.
f.issues[0].UpdatedAt = ts(3, 5, 10, "+01:00")
f.rateLimitReviewsOnce = 1
second := f.ingester(t, db)
slept := 0
second.c.sleep = func(time.Duration) { slept++ }
if err := second.repo(); err != nil {
t.Fatal(err)
}
if err := second.issues(); err != nil {
t.Fatal(err)
}
err := second.reviews()
if err == nil {
t.Fatal("expected a rate-limit error rather than a silent wait")
}
var rl *rateLimitError
if !errors.As(err, &rl) {
t.Errorf("error = %v, want a *rateLimitError the caller can recognise", err)
}
if slept != 0 {
t.Errorf("slept %d times on an incremental run, want 0", slept)
}
}
// The wait must be read from whichever bucket is actually exhausted (r=0),
// not the first one in the header.
func TestRateLimitResetPicksExhaustedBucket(t *testing.T) {
h := http.Header{}
h.Add("RateLimit", `"git_op";r=0;t=487`)
h.Add("RateLimit", `"baseline";r=1673;t=600`)
if got := rateLimitReset(h); got != 487*time.Second {
t.Errorf("= %s, want 487s from the exhausted git_op bucket", got)
}
// No exhausted bucket, and no header at all, both fall back rather than
// returning zero and spinning.
if got := rateLimitReset(http.Header{}); got != time.Minute {
t.Errorf("empty header = %s, want the 1m fallback", got)
}
}
// --repo must be validated before any request is made, or a typo becomes a
// confusing 404 from a URL the user never typed.
func TestRepoFlagValidation(t *testing.T) {
for _, bad := range []string{"flohmarkt", "a/b/c", "/leading", "trailing/"} {
err := runIngest([]string{"--db", t.TempDir() + "/x.db", "--repo", bad})
if err == nil {
t.Errorf("--repo %q accepted, want a validation error", bad)
continue
}
if !strings.Contains(err.Error(), "owner/name") {
t.Errorf("--repo %q: error = %v, want it to explain the expected form", bad, err)
}
}
}
// ---------------------------------------------------------------------------
// dLogin is the anonymous struct the API types use for a user reference.
type dLogin = struct {
Login string `json:"login"`
}
func ptr[T any](v T) *T { return &v }
func runAllPhases(t *testing.T, ing *ingester) {
t.Helper()
if err := ing.repo(); err != nil {
t.Fatalf("repo: %v", err)
}
if err := ing.labels(); err != nil {
t.Fatalf("labels: %v", err)
}
if err := ing.issues(); err != nil {
t.Fatalf("issues: %v", err)
}
if err := ing.pulls(); err != nil {
t.Fatalf("pulls: %v", err)
}
if err := ing.comments(); err != nil {
t.Fatalf("comments: %v", err)
}
if err := ing.reviews(); err != nil {
t.Fatalf("reviews: %v", err)
}
}
// snapshot reduces the database to a comparable string, so the idempotency
// test compares contents rather than just row counts.
func snapshot(t *testing.T, db *sql.DB) string {
t.Helper()
var b strings.Builder
for _, q := range []string{
`SELECT number, id, is_pull, title, state, author, updated_at FROM issue ORDER BY number`,
`SELECT number, merged, additions, deletions FROM pull ORDER BY number`,
`SELECT id, number, author, body FROM comment ORDER BY id`,
`SELECT id, number, state, author FROM review ORDER BY id`,
`SELECT id, review_id, path, body FROM review_comment ORDER BY id`,
`SELECT repo_id, number, label_id FROM issue_label ORDER BY number, label_id`,
} {
rows, err := db.Query(q)
if err != nil {
t.Fatalf("%s: %v", q, err)
}
cols, err := rows.Columns()
if err != nil {
t.Fatal(err)
}
for rows.Next() {
vals := make([]any, len(cols))
ptrs := make([]any, len(cols))
for i := range vals {
ptrs[i] = &vals[i]
}
if err := rows.Scan(ptrs...); err != nil {
t.Fatal(err)
}
fmt.Fprintf(&b, "%v\n", vals)
}
rows.Close()
b.WriteString("--\n")
}
return b.String()
}
// wrapHandler lets a test observe the requests made without reimplementing
// the fake.
func wrapHandler(h http.Handler, observe func(*url.URL)) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
observe(r.URL)
h.ServeHTTP(w, r)
})
}
|