diff --git a/internal/fetch/runtime_e2e_test.go b/internal/fetch/runtime_e2e_test.go new file mode 100644 index 0000000..4bfe80b --- /dev/null +++ b/internal/fetch/runtime_e2e_test.go @@ -0,0 +1,620 @@ +package fetch + +import ( + "context" + "net/http" + "strings" + "testing" + "time" + + "github.com/CrisSTEM/signalscope/internal/config" + "github.com/CrisSTEM/signalscope/internal/storage" +) + +func TestRuntimeExecuteEndToEndPersistsRowsAcrossSourcesAndReruns(t *testing.T) { + t.Parallel() + + const ( + datasetID = "runtime-ingestion-e2e-success" + changelogBindingID = "changelog-binding-rss-1" + ) + + ctx := context.Background() + + githubCapturedAt := time.Date(2026, 4, 23, 10, 0, 0, 0, time.UTC) + newsFirstCapturedAt := time.Date(2026, 4, 23, 11, 0, 0, 0, time.UTC) + newsSecondCapturedAt := time.Date(2026, 4, 23, 12, 0, 0, 0, time.UTC) + changelogFirstCapturedAt := time.Date(2026, 4, 23, 13, 0, 0, 0, time.UTC) + changelogSecondCapturedAt := time.Date(2026, 4, 23, 14, 0, 0, 0, time.UTC) + + githubServer := newGitHubTestServer(t, githubCapturedAt, githubTestServerOptions{}) + defer githubServer.Close() + + newsServer := newNewsRSSTestServer(t, newsRSSTestServerOptions{ + WantQuery: "Example Org", + Responses: []newsRSSTestResponse{ + { + Status: http.StatusOK, + Body: runtimeIngestionNewsRSSFirstResponse(), + }, + { + Status: http.StatusOK, + Body: runtimeIngestionNewsRSSSecondResponse(), + }, + }, + }) + defer newsServer.Close() + + changelogServer := newChangelogHTTPTestServer(t, changelogHTTPTestServerOptions{ + Responses: map[string][]changelogHTTPResponse{ + "/feeds/changelog.xml": { + { + ContentType: "application/rss+xml; charset=utf-8", + Body: runtimeIngestionChangelogRSSFirstResponse(), + }, + { + ContentType: "application/rss+xml; charset=utf-8", + Body: runtimeIngestionChangelogRSSSecondResponse(), + }, + }, + }, + }) + defer changelogServer.Close() + + pack := runtimeIngestionFixturePack( + datasetID, + changelogBindingID, + changelogServer.URL+"/feeds/changelog.xml", + ) + + db := openRuntimeTestDB(t, pack) + defer db.Close() + + registry := NewRegistry() + registry.MustRegister(SourceKindGitHub, newGitHubTestFetcher(githubServer)) + registry.MustRegister(SourceKindNewsRSS, newNewsRSSTestFetcher(newsServer)) + registry.MustRegister(SourceKindChangelog, newChangelogTestFetcher(changelogServer)) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + githubCapturedAt, + githubCapturedAt.Add(time.Second), + newsFirstCapturedAt, + newsFirstCapturedAt.Add(time.Second), + newsSecondCapturedAt, + newsSecondCapturedAt.Add(time.Second), + changelogFirstCapturedAt, + changelogFirstCapturedAt.Add(time.Second), + changelogSecondCapturedAt, + changelogSecondCapturedAt.Add(time.Second), + ), + } + + githubSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindGitHub, + BindingID: "github-binding-1", + }) + if err != nil { + t.Fatalf("Execute(github) error = %v", err) + } + if githubSummary.Attempted != 1 { + t.Fatalf("githubSummary.Attempted = %d, want %d", githubSummary.Attempted, 1) + } + if githubSummary.Succeeded != 1 { + t.Fatalf("githubSummary.Succeeded = %d, want %d", githubSummary.Succeeded, 1) + } + if githubSummary.Failed != 0 { + t.Fatalf("githubSummary.Failed = %d, want %d", githubSummary.Failed, 0) + } + if githubSummary.RecordsWritten != 1 { + t.Fatalf("githubSummary.RecordsWritten = %d, want %d", githubSummary.RecordsWritten, 1) + } + if githubSummary.MetricsWritten != 7 { + t.Fatalf("githubSummary.MetricsWritten = %d, want %d", githubSummary.MetricsWritten, 7) + } + if githubSummary.ContentItemsWritten != 0 { + t.Fatalf("githubSummary.ContentItemsWritten = %d, want %d", githubSummary.ContentItemsWritten, 0) + } + + newsFirstSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindNewsRSS, + BindingID: "news-binding-1", + }) + if err != nil { + t.Fatalf("Execute(news first run) error = %v", err) + } + if newsFirstSummary.Attempted != 1 { + t.Fatalf("newsFirstSummary.Attempted = %d, want %d", newsFirstSummary.Attempted, 1) + } + if newsFirstSummary.Succeeded != 1 { + t.Fatalf("newsFirstSummary.Succeeded = %d, want %d", newsFirstSummary.Succeeded, 1) + } + if newsFirstSummary.Failed != 0 { + t.Fatalf("newsFirstSummary.Failed = %d, want %d", newsFirstSummary.Failed, 0) + } + if newsFirstSummary.RecordsWritten != 2 { + t.Fatalf("newsFirstSummary.RecordsWritten = %d, want %d", newsFirstSummary.RecordsWritten, 2) + } + if newsFirstSummary.MetricsWritten != 1 { + t.Fatalf("newsFirstSummary.MetricsWritten = %d, want %d", newsFirstSummary.MetricsWritten, 1) + } + if newsFirstSummary.ContentItemsWritten != 2 { + t.Fatalf("newsFirstSummary.ContentItemsWritten = %d, want %d", newsFirstSummary.ContentItemsWritten, 2) + } + + newsSecondSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindNewsRSS, + BindingID: "news-binding-1", + }) + if err != nil { + t.Fatalf("Execute(news second run) error = %v", err) + } + if newsSecondSummary.Attempted != 1 { + t.Fatalf("newsSecondSummary.Attempted = %d, want %d", newsSecondSummary.Attempted, 1) + } + if newsSecondSummary.Succeeded != 1 { + t.Fatalf("newsSecondSummary.Succeeded = %d, want %d", newsSecondSummary.Succeeded, 1) + } + if newsSecondSummary.Failed != 0 { + t.Fatalf("newsSecondSummary.Failed = %d, want %d", newsSecondSummary.Failed, 0) + } + if newsSecondSummary.RecordsWritten != 2 { + t.Fatalf("newsSecondSummary.RecordsWritten = %d, want %d", newsSecondSummary.RecordsWritten, 2) + } + if newsSecondSummary.MetricsWritten != 1 { + t.Fatalf("newsSecondSummary.MetricsWritten = %d, want %d", newsSecondSummary.MetricsWritten, 1) + } + if newsSecondSummary.ContentItemsWritten != 0 { + t.Fatalf("newsSecondSummary.ContentItemsWritten = %d, want %d", newsSecondSummary.ContentItemsWritten, 0) + } + + changelogFirstSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindChangelog, + BindingID: changelogBindingID, + }) + if err != nil { + t.Fatalf("Execute(changelog first run) error = %v", err) + } + if changelogFirstSummary.Attempted != 1 { + t.Fatalf("changelogFirstSummary.Attempted = %d, want %d", changelogFirstSummary.Attempted, 1) + } + if changelogFirstSummary.Succeeded != 1 { + t.Fatalf("changelogFirstSummary.Succeeded = %d, want %d", changelogFirstSummary.Succeeded, 1) + } + if changelogFirstSummary.Failed != 0 { + t.Fatalf("changelogFirstSummary.Failed = %d, want %d", changelogFirstSummary.Failed, 0) + } + if changelogFirstSummary.RecordsWritten != 2 { + t.Fatalf("changelogFirstSummary.RecordsWritten = %d, want %d", changelogFirstSummary.RecordsWritten, 2) + } + if changelogFirstSummary.MetricsWritten != 0 { + t.Fatalf("changelogFirstSummary.MetricsWritten = %d, want %d", changelogFirstSummary.MetricsWritten, 0) + } + if changelogFirstSummary.ContentItemsWritten != 2 { + t.Fatalf("changelogFirstSummary.ContentItemsWritten = %d, want %d", changelogFirstSummary.ContentItemsWritten, 2) + } + + changelogSecondSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindChangelog, + BindingID: changelogBindingID, + }) + if err != nil { + t.Fatalf("Execute(changelog second run) error = %v", err) + } + if changelogSecondSummary.Attempted != 1 { + t.Fatalf("changelogSecondSummary.Attempted = %d, want %d", changelogSecondSummary.Attempted, 1) + } + if changelogSecondSummary.Succeeded != 1 { + t.Fatalf("changelogSecondSummary.Succeeded = %d, want %d", changelogSecondSummary.Succeeded, 1) + } + if changelogSecondSummary.Failed != 0 { + t.Fatalf("changelogSecondSummary.Failed = %d, want %d", changelogSecondSummary.Failed, 0) + } + if changelogSecondSummary.RecordsWritten != 2 { + t.Fatalf("changelogSecondSummary.RecordsWritten = %d, want %d", changelogSecondSummary.RecordsWritten, 2) + } + if changelogSecondSummary.MetricsWritten != 0 { + t.Fatalf("changelogSecondSummary.MetricsWritten = %d, want %d", changelogSecondSummary.MetricsWritten, 0) + } + if changelogSecondSummary.ContentItemsWritten != 0 { + t.Fatalf("changelogSecondSummary.ContentItemsWritten = %d, want %d", changelogSecondSummary.ContentItemsWritten, 0) + } + + fetchRunCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM fetch_runs WHERE dataset_id = ?`, datasetID) + if err != nil { + t.Fatalf("queryCount(fetch_runs) error = %v", err) + } + if fetchRunCount != 5 { + t.Fatalf("fetchRunCount = %d, want %d", fetchRunCount, 5) + } + + succeededRunCount, err := queryCount( + ctx, + db, + `SELECT COUNT(*) FROM fetch_runs WHERE dataset_id = ? AND status = ?`, + datasetID, + storage.FetchRunStatusSucceeded, + ) + if err != nil { + t.Fatalf("queryCount(fetch_runs succeeded) error = %v", err) + } + if succeededRunCount != 5 { + t.Fatalf("succeededRunCount = %d, want %d", succeededRunCount, 5) + } + + metricCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM metric_snapshots WHERE dataset_id = ?`, datasetID) + if err != nil { + t.Fatalf("queryCount(metric_snapshots) error = %v", err) + } + if metricCount != 9 { + t.Fatalf("metricCount = %d, want %d", metricCount, 9) + } + + contentCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM content_items WHERE dataset_id = ?`, datasetID) + if err != nil { + t.Fatalf("queryCount(content_items) error = %v", err) + } + if contentCount != 4 { + t.Fatalf("contentCount = %d, want %d", contentCount, 4) + } + + githubRows, err := readMetricSnapshotsForBinding(ctx, db, "github-binding-1") + if err != nil { + t.Fatalf("readMetricSnapshotsForBinding() error = %v", err) + } + if len(githubRows) != 7 { + t.Fatalf("len(githubRows) = %d, want %d", len(githubRows), 7) + } + + newsSnapshotRows, err := readNewsMetricSnapshotsForBinding(ctx, db, "news-binding-1") + if err != nil { + t.Fatalf("readNewsMetricSnapshotsForBinding() error = %v", err) + } + if len(newsSnapshotRows) != 2 { + t.Fatalf("len(newsSnapshotRows) = %d, want %d", len(newsSnapshotRows), 2) + } + if !newsSnapshotRows[0].MetricValueNum.Valid || int(newsSnapshotRows[0].MetricValueNum.Float64) != 2 { + t.Fatalf("newsSnapshotRows[0].MetricValueNum = %+v, want %d", newsSnapshotRows[0].MetricValueNum, 2) + } + if !newsSnapshotRows[1].MetricValueNum.Valid || int(newsSnapshotRows[1].MetricValueNum.Float64) != 2 { + t.Fatalf("newsSnapshotRows[1].MetricValueNum = %+v, want %d", newsSnapshotRows[1].MetricValueNum, 2) + } + if newsSnapshotRows[0].CapturedAt != newsFirstCapturedAt.Format(time.RFC3339) { + t.Fatalf("newsSnapshotRows[0].CapturedAt = %q, want %q", newsSnapshotRows[0].CapturedAt, newsFirstCapturedAt.Format(time.RFC3339)) + } + if newsSnapshotRows[1].CapturedAt != newsSecondCapturedAt.Format(time.RFC3339) { + t.Fatalf("newsSnapshotRows[1].CapturedAt = %q, want %q", newsSnapshotRows[1].CapturedAt, newsSecondCapturedAt.Format(time.RFC3339)) + } + + newsContentRows, err := readNewsContentItemsForBinding(ctx, db, "news-binding-1") + if err != nil { + t.Fatalf("readNewsContentItemsForBinding() error = %v", err) + } + if len(newsContentRows) != 2 { + t.Fatalf("len(newsContentRows) = %d, want %d", len(newsContentRows), 2) + } + if newsContentRows[0].Title != "Example Org launches satellite" { + t.Fatalf("newsContentRows[0].Title = %q, want %q", newsContentRows[0].Title, "Example Org launches satellite") + } + if newsContentRows[0].Summary != "Original summary" { + t.Fatalf("newsContentRows[0].Summary = %q, want %q", newsContentRows[0].Summary, "Original summary") + } + if newsContentRows[0].DiscoveredAt != newsFirstCapturedAt.Format(time.RFC3339) { + t.Fatalf("newsContentRows[0].DiscoveredAt = %q, want %q", newsContentRows[0].DiscoveredAt, newsFirstCapturedAt.Format(time.RFC3339)) + } + + changelogRows, err := readChangelogContentItemsForBinding(ctx, db, changelogBindingID) + if err != nil { + t.Fatalf("readChangelogContentItemsForBinding() error = %v", err) + } + if len(changelogRows) != 2 { + t.Fatalf("len(changelogRows) = %d, want %d", len(changelogRows), 2) + } + if changelogRows[0].Title != "Version 2.0 shipped" { + t.Fatalf("changelogRows[0].Title = %q, want %q", changelogRows[0].Title, "Version 2.0 shipped") + } + if changelogRows[0].Summary != "Original changelog body" { + t.Fatalf("changelogRows[0].Summary = %q, want %q", changelogRows[0].Summary, "Original changelog body") + } + if changelogRows[0].DiscoveredAt != changelogFirstCapturedAt.Format(time.RFC3339) { + t.Fatalf("changelogRows[0].DiscoveredAt = %q, want %q", changelogRows[0].DiscoveredAt, changelogFirstCapturedAt.Format(time.RFC3339)) + } + if changelogRows[1].URL != changelogServer.URL+"/posts/migration-note" { + t.Fatalf("changelogRows[1].URL = %q, want %q", changelogRows[1].URL, changelogServer.URL+"/posts/migration-note") + } +} + +func TestRuntimeExecuteEndToEndMixedSuccessFailureAcrossInvocations(t *testing.T) { + t.Parallel() + + const ( + datasetID = "runtime-ingestion-e2e-failure" + changelogBindingID = "changelog-binding-rss-1" + ) + + ctx := context.Background() + + githubCapturedAt := time.Date(2026, 4, 24, 10, 0, 0, 0, time.UTC) + newsCapturedAt := time.Date(2026, 4, 24, 11, 0, 0, 0, time.UTC) + changelogCapturedAt := time.Date(2026, 4, 24, 12, 0, 0, 0, time.UTC) + + githubServer := newGitHubTestServer(t, githubCapturedAt, githubTestServerOptions{ + Repo2Status: http.StatusInternalServerError, + }) + defer githubServer.Close() + + newsServer := newNewsRSSTestServer(t, newsRSSTestServerOptions{ + WantQuery: "Example Org", + Responses: []newsRSSTestResponse{ + { + Status: http.StatusOK, + Body: runtimeIngestionNewsRSSFirstResponse(), + }, + }, + }) + defer newsServer.Close() + + changelogServer := newChangelogHTTPTestServer(t, changelogHTTPTestServerOptions{ + Responses: map[string][]changelogHTTPResponse{ + "/feeds/changelog.xml": { + { + ContentType: "application/rss+xml; charset=utf-8", + Body: runtimeIngestionChangelogRSSFirstResponse(), + }, + }, + }, + }) + defer changelogServer.Close() + + pack := runtimeIngestionFixturePack( + datasetID, + changelogBindingID, + changelogServer.URL+"/feeds/changelog.xml", + ) + + db := openRuntimeTestDB(t, pack) + defer db.Close() + + registry := NewRegistry() + registry.MustRegister(SourceKindGitHub, newGitHubTestFetcher(githubServer)) + registry.MustRegister(SourceKindNewsRSS, newNewsRSSTestFetcher(newsServer)) + registry.MustRegister(SourceKindChangelog, newChangelogTestFetcher(changelogServer)) + + runtime := Runtime{ + DB: db, + Registry: registry, + Now: sequenceClock( + githubCapturedAt, + githubCapturedAt.Add(time.Second), + githubCapturedAt.Add(2*time.Minute), + githubCapturedAt.Add(2*time.Minute+time.Second), + newsCapturedAt, + newsCapturedAt.Add(time.Second), + changelogCapturedAt, + changelogCapturedAt.Add(time.Second), + ), + } + + githubSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindGitHub, + }) + if err != nil { + t.Fatalf("Execute(github) error = %v", err) + } + if githubSummary.Attempted != 2 { + t.Fatalf("githubSummary.Attempted = %d, want %d", githubSummary.Attempted, 2) + } + if githubSummary.Succeeded != 1 { + t.Fatalf("githubSummary.Succeeded = %d, want %d", githubSummary.Succeeded, 1) + } + if githubSummary.Failed != 1 { + t.Fatalf("githubSummary.Failed = %d, want %d", githubSummary.Failed, 1) + } + if !githubSummary.HasFailures() { + t.Fatal("githubSummary.HasFailures() = false, want true") + } + if githubSummary.RecordsWritten != 1 { + t.Fatalf("githubSummary.RecordsWritten = %d, want %d", githubSummary.RecordsWritten, 1) + } + if githubSummary.MetricsWritten != 7 { + t.Fatalf("githubSummary.MetricsWritten = %d, want %d", githubSummary.MetricsWritten, 7) + } + if githubSummary.ContentItemsWritten != 0 { + t.Fatalf("githubSummary.ContentItemsWritten = %d, want %d", githubSummary.ContentItemsWritten, 0) + } + + githubFailedResult := mustFindBindingResult(t, githubSummary.Results, "github-binding-2") + if githubFailedResult.Status != storage.FetchRunStatusFailed { + t.Fatalf("githubFailedResult.Status = %q, want %q", githubFailedResult.Status, storage.FetchRunStatusFailed) + } + if !strings.Contains(githubFailedResult.ErrorMessage, "simulated upstream failure") { + t.Fatalf("githubFailedResult.ErrorMessage = %q, want simulated upstream failure", githubFailedResult.ErrorMessage) + } + + newsSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindNewsRSS, + BindingID: "news-binding-1", + }) + if err != nil { + t.Fatalf("Execute(news) error = %v", err) + } + if newsSummary.Attempted != 1 { + t.Fatalf("newsSummary.Attempted = %d, want %d", newsSummary.Attempted, 1) + } + if newsSummary.Succeeded != 1 { + t.Fatalf("newsSummary.Succeeded = %d, want %d", newsSummary.Succeeded, 1) + } + if newsSummary.Failed != 0 { + t.Fatalf("newsSummary.Failed = %d, want %d", newsSummary.Failed, 0) + } + if newsSummary.MetricsWritten != 1 { + t.Fatalf("newsSummary.MetricsWritten = %d, want %d", newsSummary.MetricsWritten, 1) + } + if newsSummary.ContentItemsWritten != 2 { + t.Fatalf("newsSummary.ContentItemsWritten = %d, want %d", newsSummary.ContentItemsWritten, 2) + } + + changelogSummary, err := runtime.Execute(ctx, pack, Options{ + SourceKind: SourceKindChangelog, + BindingID: changelogBindingID, + }) + if err != nil { + t.Fatalf("Execute(changelog) error = %v", err) + } + if changelogSummary.Attempted != 1 { + t.Fatalf("changelogSummary.Attempted = %d, want %d", changelogSummary.Attempted, 1) + } + if changelogSummary.Succeeded != 1 { + t.Fatalf("changelogSummary.Succeeded = %d, want %d", changelogSummary.Succeeded, 1) + } + if changelogSummary.Failed != 0 { + t.Fatalf("changelogSummary.Failed = %d, want %d", changelogSummary.Failed, 0) + } + if changelogSummary.MetricsWritten != 0 { + t.Fatalf("changelogSummary.MetricsWritten = %d, want %d", changelogSummary.MetricsWritten, 0) + } + if changelogSummary.ContentItemsWritten != 2 { + t.Fatalf("changelogSummary.ContentItemsWritten = %d, want %d", changelogSummary.ContentItemsWritten, 2) + } + + fetchRunCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM fetch_runs WHERE dataset_id = ?`, datasetID) + if err != nil { + t.Fatalf("queryCount(fetch_runs) error = %v", err) + } + if fetchRunCount != 4 { + t.Fatalf("fetchRunCount = %d, want %d", fetchRunCount, 4) + } + + succeededRunCount, err := queryCount( + ctx, + db, + `SELECT COUNT(*) FROM fetch_runs WHERE dataset_id = ? AND status = ?`, + datasetID, + storage.FetchRunStatusSucceeded, + ) + if err != nil { + t.Fatalf("queryCount(fetch_runs succeeded) error = %v", err) + } + if succeededRunCount != 3 { + t.Fatalf("succeededRunCount = %d, want %d", succeededRunCount, 3) + } + + failedRunCount, err := queryCount( + ctx, + db, + `SELECT COUNT(*) FROM fetch_runs WHERE dataset_id = ? AND status = ?`, + datasetID, + storage.FetchRunStatusFailed, + ) + if err != nil { + t.Fatalf("queryCount(fetch_runs failed) error = %v", err) + } + if failedRunCount != 1 { + t.Fatalf("failedRunCount = %d, want %d", failedRunCount, 1) + } + + metricCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM metric_snapshots WHERE dataset_id = ?`, datasetID) + if err != nil { + t.Fatalf("queryCount(metric_snapshots) error = %v", err) + } + if metricCount != 8 { + t.Fatalf("metricCount = %d, want %d", metricCount, 8) + } + + contentCount, err := queryCount(ctx, db, `SELECT COUNT(*) FROM content_items WHERE dataset_id = ?`, datasetID) + if err != nil { + t.Fatalf("queryCount(content_items) error = %v", err) + } + if contentCount != 4 { + t.Fatalf("contentCount = %d, want %d", contentCount, 4) + } + + failedBindingSnapshots, err := queryCount(ctx, db, `SELECT COUNT(*) FROM metric_snapshots WHERE binding_id = ?`, "github-binding-2") + if err != nil { + t.Fatalf("queryCount(failed binding metric_snapshots) error = %v", err) + } + if failedBindingSnapshots != 0 { + t.Fatalf("failedBindingSnapshots = %d, want %d", failedBindingSnapshots, 0) + } + + rows, err := readFetchRuns(ctx, db) + if err != nil { + t.Fatalf("readFetchRuns() error = %v", err) + } + if len(rows) != 4 { + t.Fatalf("len(rows) = %d, want %d", len(rows), 4) + } + + failedRun := mustFindFetchRunRow(t, rows, "github-binding-2") + if failedRun.Status != storage.FetchRunStatusFailed { + t.Fatalf("failedRun.Status = %q, want %q", failedRun.Status, storage.FetchRunStatusFailed) + } + if !strings.Contains(failedRun.ErrorMessage, "simulated upstream failure") { + t.Fatalf("failedRun.ErrorMessage = %q, want simulated upstream failure", failedRun.ErrorMessage) + } + + if mustFindFetchRunRow(t, rows, "github-binding-1").Status != storage.FetchRunStatusSucceeded { + t.Fatalf("github-binding-1 fetch run status = %q, want %q", mustFindFetchRunRow(t, rows, "github-binding-1").Status, storage.FetchRunStatusSucceeded) + } + if mustFindFetchRunRow(t, rows, "news-binding-1").Status != storage.FetchRunStatusSucceeded { + t.Fatalf("news-binding-1 fetch run status = %q, want %q", mustFindFetchRunRow(t, rows, "news-binding-1").Status, storage.FetchRunStatusSucceeded) + } + if mustFindFetchRunRow(t, rows, changelogBindingID).Status != storage.FetchRunStatusSucceeded { + t.Fatalf("%s fetch run status = %q, want %q", changelogBindingID, mustFindFetchRunRow(t, rows, changelogBindingID).Status, storage.FetchRunStatusSucceeded) + } +} + +func runtimeIngestionFixturePack(datasetID, changelogBindingID, changelogFeedURL string) config.Pack { + return runtimeFixturePackWithChangelogBinding( + datasetID, + changelogBindingID, + config.JSONMap{ + "mode": "rss", + "feed_url": changelogFeedURL, + }, + ) +} + +func mustFindBindingResult(t *testing.T, results []BindingResult, bindingID string) BindingResult { + t.Helper() + + for _, result := range results { + if result.BindingID == bindingID { + return result + } + } + + t.Fatalf("binding result %q not found", bindingID) + return BindingResult{} +} + +func mustFindFetchRunRow(t *testing.T, rows []fetchRunRow, bindingID string) fetchRunRow { + t.Helper() + + for _, row := range rows { + if row.BindingID == bindingID { + return row + } + } + + t.Fatalf("fetch run row %q not found", bindingID) + return fetchRunRow{} +} + +func runtimeIngestionNewsRSSFirstResponse() string { + return `Example Org launches satellitehttps://example.com/article-1Original summaryguid-1Tue, 21 Apr 2026 10:30:00 GMTExample Org hires new CTOhttps://example.com/article-2Original second summaryTue, 21 Apr 2026 10:45:00 GMT` +} + +func runtimeIngestionNewsRSSSecondResponse() string { + return `Example Org launches satellite UPDATEDhttps://example.com/article-1Updated summary should not overwrite the first rowguid-1Tue, 21 Apr 2026 10:30:00 GMTExample Org hires new CTOhttps://example.com/article-2Updated second summary should not overwrite the first rowTue, 21 Apr 2026 10:45:00 GMT` +} + +func runtimeIngestionChangelogRSSFirstResponse() string { + return `Example ChangelogVersion 2.0 shippedhttps://example.com/changelog/v2Original changelog bodyguid-1Tue, 22 Apr 2026 12:30:00 GMTMigration note published/posts/migration-noteSecond changelog entryTue, 22 Apr 2026 12:45:00 GMT` +} + +func runtimeIngestionChangelogRSSSecondResponse() string { + return `Example ChangelogVersion 2.0 shipped UPDATEDhttps://example.com/changelog/v2Updated body should not overwrite first rowguid-1Tue, 22 Apr 2026 12:30:00 GMTMigration note published UPDATED/posts/migration-noteUpdated second entry should not overwrite first rowTue, 22 Apr 2026 12:45:00 GMT` +}