eric/weatherreporter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Upload batch distributor notifications

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-17 21:00:03 +00:00
parent 1d2f176977
commit 2ef91cf1b1
4 changed files with 563 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

298
internal/app/app_test.go
View File

@@ -2447,10 +2447,11 @@ func TestBuildBatchNotificationRequestRejectsMissingReportPath(t *testing.T) {
func TestRunBatchContinuesAfterReportFailure(t *testing.T) {
server := dailyBundleServer(t)
cfg := dailyWorkspaceConfig(t, server)
cfg := dailyNotificationConfig(t, server)
collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31")
cfg.WeatherAPI.BaseURL = ""
collector := &recordingCollector{result: &collection}
notifier := &recordingNotifier{}
renderer := &selectiveRenderer{
failRenderPrompt: "weather.tomorrow_generated_text",
runBody: "# Batch Report\n",
@@ -2462,6 +2463,7 @@ func TestRunBatchContinuesAfterReportFailure(t *testing.T) {
Now: mustParse("2026-05-29T05:00:00-05:00"),
Collector: collector,
Renderer: renderer,
Notifier: notifier,
})
if err != nil {
t.Fatalf("RunBatchDetailed() error = %v", err)
@@ -2476,6 +2478,12 @@ func TestRunBatchContinuesAfterReportFailure(t *testing.T) {
if renderer.runCalls != 0 || renderer.structuredRunCalls != 2 {
t.Fatalf("renderer calls run=%d structured=%d, want successful reports to continue", renderer.runCalls, renderer.structuredRunCalls)
}
if len(notifier.requests) != 0 || len(notifier.batchRequests) != 0 {
t.Fatalf("notification requests report=%d batch=%d, want none after report failure", len(notifier.requests), len(notifier.batchRequests))
}
if result.Notification == nil || result.Notification.Status != "skipped" || result.Notification.Reason != "one or more reports failed" {
t.Fatalf("batch notification = %#v, want skipped after report failure", result.Notification)
}
var failedTomorrow bool
var succeededDaily bool
for _, item := range result.Reports {
@@ -2497,6 +2505,64 @@ func TestRunBatchContinuesAfterReportFailure(t *testing.T) {
}
}
func TestRunBatchMorningSendsOneBatchNotification(t *testing.T) {
server := dailyBundleServer(t)
cfg := dailyNotificationConfig(t, server)
collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31")
cfg.WeatherAPI.BaseURL = ""
collector := &recordingCollector{result: &collection}
notifier := &recordingNotifier{batchResult: successfulBatchNotificationResult()}
result, err := RunBatchDetailed(context.Background(), BatchRequest{
Config: cfg,
Batch: BatchMorning,
Now: mustParse("2026-05-29T05:00:00-05:00"),
Collector: collector,
Renderer: &selectiveRenderer{runBody: "# Batch Report\n"},
Notifier: notifier,
})
if err != nil {
t.Fatalf("RunBatchDetailed() error = %v", err)
}
if result.Total != 3 || result.Succeeded != 3 || result.Failed != 0 {
t.Fatalf("summary total/succeeded/failed = %d/%d/%d, want 3/3/0", result.Total, result.Succeeded, result.Failed)
}
if len(notifier.requests) != 0 {
t.Fatalf("per-report notification requests = %#v, want none", notifier.requests)
}
if len(notifier.batchRequests) != 1 {
t.Fatalf("batch notification requests = %d, want 1", len(notifier.batchRequests))
}
req := notifier.batchRequests[0]
if req.Batch != BatchMorning || req.RunID != "20260529T100000.000000000Z_morning" {
t.Fatalf("batch request identity = %s/%s, want morning run id", req.Batch, req.RunID)
}
if len(req.IncludedReports) != 3 || len(req.Files) != 3 {
t.Fatalf("batch request reports/files = %d/%d, want 3/3", len(req.IncludedReports), len(req.Files))
}
for _, file := range req.Files {
if file.SourcePath == "" || file.BundlePath == "" {
t.Fatalf("batch file = %#v, want source and bundle path", file)
}
if !strings.Contains(file.BundlePath, file.RunID) {
t.Fatalf("bundle path %q does not include report run id %q", file.BundlePath, file.RunID)
}
}
if result.Notification == nil || result.Notification.Status != "succeeded" || result.Notification.RunID != "batch-distributor-run" || result.Notification.Path == "" {
t.Fatalf("batch notification = %#v, want succeeded result with artifact path", result.Notification)
}
if len(result.Notification.IncludedReports) != 3 {
t.Fatalf("batch notification included reports = %d, want 3", len(result.Notification.IncludedReports))
}
artifact := readBatchNotificationForTest(t, result.Notification.Path)
if artifact.Status != "succeeded" || artifact.Upload == nil || artifact.Upload.RunID != "batch-distributor-run" || artifact.RunStatus == nil {
t.Fatalf("batch notification artifact = %#v, want succeeded upload and run status", artifact)
}
if len(artifact.Reports) != 3 {
t.Fatalf("artifact included reports = %d, want 3", len(artifact.Reports))
}
}
func TestRunBatchSuppressesPerReportNotification(t *testing.T) {
server := dailyBundleServer(t)
cfg := dailyNotificationConfig(t, server)
@@ -2528,6 +2594,9 @@ func TestRunBatchSuppressesPerReportNotification(t *testing.T) {
if len(notifier.requests) != 0 {
t.Fatalf("notification requests = %#v, want none for batch-generated reports", notifier.requests)
}
if len(notifier.batchRequests) != 1 {
t.Fatalf("batch notification requests = %d, want 1", len(notifier.batchRequests))
}
for _, item := range result.Reports {
if item.Status != "succeeded" {
t.Fatalf("report %s status = %s, want succeeded", item.ReportID, item.Status)
@@ -2540,11 +2609,152 @@ func TestRunBatchSuppressesPerReportNotification(t *testing.T) {
t.Fatalf("report %s metadata NotificationPath = %q, want empty", item.ReportID, metadata.NotificationPath)
}
}
notificationDir := filepath.Join(cfg.Workspace.Root, cfg.Workspace.NotificationsDir)
if _, err := os.Stat(notificationDir); err == nil {
t.Fatalf("notification directory %q exists, want no per-report notification artifacts", notificationDir)
} else if !os.IsNotExist(err) {
t.Fatalf("stat notification directory %q: %v", notificationDir, err)
for _, item := range result.Reports {
reportNotificationDir := filepath.Join(cfg.Workspace.Root, cfg.Workspace.NotificationsDir, string(item.ReportID))
if _, err := os.Stat(reportNotificationDir); err == nil {
t.Fatalf("per-report notification directory %q exists, want none", reportNotificationDir)
} else if !os.IsNotExist(err) {
t.Fatalf("stat per-report notification directory %q: %v", reportNotificationDir, err)
}
}
}
func TestRunBatchNotificationFailureKeepsReportItemsSucceeded(t *testing.T) {
server := dailyBundleServer(t)
cfg := dailyNotificationConfig(t, server)
collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31")
cfg.WeatherAPI.BaseURL = ""
notifier := &recordingNotifier{batchErr: errors.New("batch upload rejected")}
result, err := RunBatchDetailed(context.Background(), BatchRequest{
Config: cfg,
Batch: BatchEvening,
Now: mustParse("2026-05-29T18:00:00-05:00"),
Collector: &recordingCollector{result: &collection},
Renderer: &selectiveRenderer{runBody: "# Batch Report\n"},
Notifier: notifier,
})
if err != nil {
t.Fatalf("RunBatchDetailed() error = %v", err)
}
if result.Failed != 1 || result.Succeeded != 2 {
t.Fatalf("summary succeeded/failed = %d/%d, want report successes plus notification failure", result.Succeeded, result.Failed)
}
for _, item := range result.Reports {
if item.Status != "succeeded" {
t.Fatalf("report %s status = %s, want succeeded despite batch notification failure", item.ReportID, item.Status)
}
}
if result.Notification == nil || result.Notification.Status != "failed" || !strings.Contains(result.Notification.Error, "batch upload rejected") || result.Notification.Path == "" {
t.Fatalf("batch notification = %#v, want failed upload result", result.Notification)
}
artifact := readBatchNotificationForTest(t, result.Notification.Path)
if artifact.Status != "failed" || !strings.Contains(artifact.Error, "batch upload rejected") {
t.Fatalf("batch notification artifact = %#v, want failed upload error", artifact)
}
err = RunBatch(context.Background(), BatchRequest{
Config: cfg,
Batch: BatchEvening,
Now: mustParse("2026-05-29T18:00:00-05:00"),
Collector: &recordingCollector{result: &collection},
Renderer: &selectiveRenderer{runBody: "# Batch Report\n"},
Notifier: &recordingNotifier{batchErr: errors.New("batch upload rejected")},
})
var batchErr BatchError
if !errors.As(err, &batchErr) {
t.Fatalf("RunBatch() error = %T %v, want BatchError", err, err)
}
if batchErr.Result == nil || !batchNotificationFailed(batchErr.Result) || batchReportFailures(batchErr.Result) != 0 {
t.Fatalf("RunBatch() result = %#v, want notification-only batch failure", batchErr.Result)
}
}
func TestRunBatchNotificationStatusErrorPersistsStatusReport(t *testing.T) {
server := dailyBundleServer(t)
cfg := dailyNotificationConfig(t, server)
collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31")
cfg.WeatherAPI.BaseURL = ""
notifier := &recordingNotifier{
batchResult: &NotificationResult{
RunID: "batch-distributor-run",
Status: "accepted",
UploadStatus: "accepted",
StatusError: "status lookup unavailable",
Report: []byte(`{"actions":[{"action":"replace_older"}]}`),
},
}
result, err := RunBatchDetailed(context.Background(), BatchRequest{
Config: cfg,
Batch: BatchMorning,
Now: mustParse("2026-05-29T05:00:00-05:00"),
Collector: &recordingCollector{result: &collection},
Renderer: &selectiveRenderer{runBody: "# Batch Report\n"},
Notifier: notifier,
})
if err != nil {
t.Fatalf("RunBatchDetailed() error = %v", err)
}
if result.Failed != 0 || result.Notification == nil || result.Notification.Path == "" {
t.Fatalf("result = %#v, want status-error notification artifact without batch failure", result)
}
artifact := readBatchNotificationForTest(t, result.Notification.Path)
if artifact.StatusError != "status lookup unavailable" || artifact.RunStatus == nil || !strings.Contains(string(artifact.RunStatus.Report), "replace_older") {
t.Fatalf("batch notification artifact = %#v, want status error and raw status report", artifact)
}
}
func TestRunBatchDisabledDistributorSkipsBatchNotification(t *testing.T) {
server := dailyBundleServer(t)
cfg := dailyWorkspaceConfig(t, server)
collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31")
cfg.WeatherAPI.BaseURL = ""
notifier := &recordingNotifier{}
result, err := RunBatchDetailed(context.Background(), BatchRequest{
Config: cfg,
Batch: BatchEvening,
Now: mustParse("2026-05-29T18:00:00-05:00"),
Collector: &recordingCollector{result: &collection},
Renderer: &selectiveRenderer{runBody: "# Batch Report\n"},
Notifier: notifier,
})
if err != nil {
t.Fatalf("RunBatchDetailed() error = %v", err)
}
if result.Notification != nil || result.Failed != 0 {
t.Fatalf("result notification/failed = %#v/%d, want disabled notification omitted", result.Notification, result.Failed)
}
if len(notifier.requests) != 0 || len(notifier.batchRequests) != 0 {
t.Fatalf("notification requests report=%d batch=%d, want none when distributor disabled", len(notifier.requests), len(notifier.batchRequests))
}
}
func TestRunBatchDisabledBatchNotificationSkipsNotifier(t *testing.T) {
server := dailyBundleServer(t)
cfg := dailyNotificationConfig(t, server)
cfg.Notify.Distributor.Batch.Enabled = false
collection := collectionWithFutureDailyForTest(t, cfg, "2026-05-31")
cfg.WeatherAPI.BaseURL = ""
notifier := &recordingNotifier{}
result, err := RunBatchDetailed(context.Background(), BatchRequest{
Config: cfg,
Batch: BatchEvening,
Now: mustParse("2026-05-29T18:00:00-05:00"),
Collector: &recordingCollector{result: &collection},
Renderer: &selectiveRenderer{runBody: "# Batch Report\n"},
Notifier: notifier,
})
if err != nil {
t.Fatalf("RunBatchDetailed() error = %v", err)
}
if result.Notification != nil || result.Failed != 0 {
t.Fatalf("result notification/failed = %#v/%d, want disabled batch notification omitted", result.Notification, result.Failed)
}
if len(notifier.requests) != 0 || len(notifier.batchRequests) != 0 {
t.Fatalf("notification requests report=%d batch=%d, want none when batch notification disabled", len(notifier.requests), len(notifier.batchRequests))
}
}
@@ -2629,6 +2839,9 @@ func TestRunBatchDynamicDailyReportsHaveDistinctIdentity(t *testing.T) {
if len(notifier.requests) != 0 {
t.Fatalf("notification requests = %#v, want none for batch-generated reports", notifier.requests)
}
if len(notifier.batchRequests) != 1 {
t.Fatalf("batch notification requests = %d, want 1", len(notifier.batchRequests))
}
dailyByDate := map[string]BatchReportResult{}
runIDs := map[string]struct{}{}
@@ -2675,6 +2888,18 @@ func TestRunBatchDynamicDailyReportsHaveDistinctIdentity(t *testing.T) {
}
}
if len(notifier.batchRequests[0].IncludedReports) != len(result.Reports) {
t.Fatalf("batch included reports = %d, want %d", len(notifier.batchRequests[0].IncludedReports), len(result.Reports))
}
for _, included := range notifier.batchRequests[0].IncludedReports {
if strings.Contains(included.SourcePath, outputDir) {
t.Fatalf("batch notification source path = %q, want managed report path outside output dir", included.SourcePath)
}
if _, ok := reportPaths[included.SourcePath]; !ok {
t.Fatalf("batch notification source path = %q, want one of %#v", included.SourcePath, reportPaths)
}
}
}
func TestRunBatchMorningUsesTodayOutputName(t *testing.T) {
@@ -2842,6 +3067,15 @@ func successfulNotificationResult() *NotificationResult {
}
}
func successfulBatchNotificationResult() *NotificationResult {
return &NotificationResult{
RunID: "batch-distributor-run",
Status: "succeeded",
UploadStatus: "accepted",
Report: []byte(`{"actions":[{"action":"replace_older"}]}`),
}
}
func successfulGeneratedTextRenderer(body string) *recordingRenderer {
return &recordingRenderer{
renderResult: &scriptorium.RenderResult{ExitCode: 0},
@@ -3125,6 +3359,19 @@ func readMetadataForTest(t *testing.T, path string) state.Metadata {
return metadata
}
func readBatchNotificationForTest(t *testing.T, path string) state.BatchDistributorNotificationArtifact {
t.Helper()
data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read batch notification %q: %v", path, err)
}
var artifact state.BatchDistributorNotificationArtifact
if err := json.Unmarshal(data, &artifact); err != nil {
t.Fatalf("decode batch notification %q: %v", path, err)
}
return artifact
}
func assertGeneratedReportError(t *testing.T, err error, resolved report.Resolved, operation string) {
t.Helper()
if err == nil {
@@ -3337,10 +3584,13 @@ type selectiveRenderer struct {
}
type recordingNotifier struct {
requests []NotificationRequest
result *NotificationResult
err error
errByReport map[report.ID]error
requests []NotificationRequest
batchRequests []batchNotificationRequest
result *NotificationResult
batchResult *NotificationResult
err error
batchErr error
errByReport map[report.ID]error
}
func (n *recordingNotifier) Notify(_ context.Context, req NotificationRequest) (*NotificationResult, error) {
@@ -3373,6 +3623,34 @@ func (n *recordingNotifier) Notify(_ context.Context, req NotificationRequest) (
}, nil
}
func (n *recordingNotifier) NotifyBatch(_ context.Context, req batchNotificationRequest) (*NotificationResult, error) {
n.batchRequests = append(n.batchRequests, req)
if n.batchErr != nil {
return nil, n.batchErr
}
if n.batchResult != nil {
result := *n.batchResult
if result.BundleID == "" {
result.BundleID = req.BundleID
}
if result.IdempotencyKey == "" {
result.IdempotencyKey = req.IdempotencyKey
}
if result.PipelineID == "" {
result.PipelineID = req.PipelineID
}
return &result, nil
}
return &NotificationResult{
PipelineID: req.PipelineID,
BundleID: req.BundleID,
IdempotencyKey: req.IdempotencyKey,
RunID: "batch-distributor-run",
Status: "accepted",
UploadStatus: "accepted",
}, nil
}
func (r *selectiveRenderer) Render(_ context.Context, req scriptorium.RenderRequest) (*scriptorium.RenderResult, error) {
r.renderCalls++
if req.PromptID == r.failRenderPrompt {