ingest: add runtime and control ingest stats tests

internal/control/control_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	cfgpkg "github.com/jan/fm-rds-tx/internal/config"
+	"github.com/jan/fm-rds-tx/internal/ingest"
 	"github.com/jan/fm-rds-tx/internal/output"
 )
 
@@ -176,6 +177,36 @@ func TestRuntimeWithoutDriver(t *testing.T) {
 	}
 }
 
+func TestRuntimeIncludesIngestStats(t *testing.T) {
+	srv := NewServer(cfgpkg.Default())
+	srv.SetIngestRuntime(&fakeIngestRuntime{
+		stats: ingest.Stats{
+			Active:  ingest.SourceDescriptor{ID: "stdin-main", Kind: "stdin-pcm"},
+			Runtime: ingest.RuntimeStats{State: "running"},
+		},
+	})
+	rec := httptest.NewRecorder()
+	srv.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/runtime", nil))
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status: %d", rec.Code)
+	}
+	var body map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("unmarshal runtime: %v", err)
+	}
+	ingest, ok := body["ingest"].(map[string]any)
+	if !ok {
+		t.Fatalf("expected ingest stats, got %T", body["ingest"])
+	}
+	active, ok := ingest["active"].(map[string]any)
+	if !ok {
+		t.Fatalf("expected ingest.active map, got %T", ingest["active"])
+	}
+	if active["id"] != "stdin-main" {
+		t.Fatalf("unexpected ingest active id: %v", active["id"])
+	}
+}
+
 func TestRuntimeReportsFaultHistory(t *testing.T) {
 	srv := NewServer(cfgpkg.Default())
 	history := []map[string]any{
@@ -604,12 +635,20 @@ type fakeAudioIngress struct {
 	totalFrames int
 }
 
+type fakeIngestRuntime struct {
+	stats ingest.Stats
+}
+
 func (f *fakeAudioIngress) WritePCM16(data []byte) (int, error) {
 	frames := len(data) / 4
 	f.totalFrames += frames
 	return frames, nil
 }
 
+func (f *fakeIngestRuntime) Stats() ingest.Stats {
+	return f.stats
+}
+
 func (f *fakeTXController) StartTX() error { return nil }
 func (f *fakeTXController) StopTX() error  { return nil }
 func (f *fakeTXController) TXStats() map[string]any {

internal/ingest/runtime_test.go

@@ -0,0 +1,56 @@
+package ingest
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/jan/fm-rds-tx/internal/audio"
+)
+
+type fakeSource struct {
+	desc   SourceDescriptor
+	chunks chan PCMChunk
+	errs   chan error
+	stats  SourceStats
+}
+
+func newFakeSource() *fakeSource {
+	return &fakeSource{
+		desc:   SourceDescriptor{ID: "fake", Kind: "stdin-pcm"},
+		chunks: make(chan PCMChunk, 4),
+		errs:   make(chan error, 1),
+		stats:  SourceStats{State: "running", Connected: true},
+	}
+}
+
+func (s *fakeSource) Descriptor() SourceDescriptor { return s.desc }
+func (s *fakeSource) Start(context.Context) error  { return nil }
+func (s *fakeSource) Stop() error                  { close(s.chunks); return nil }
+func (s *fakeSource) Chunks() <-chan PCMChunk      { return s.chunks }
+func (s *fakeSource) Errors() <-chan error         { return s.errs }
+func (s *fakeSource) Stats() SourceStats           { return s.stats }
+
+func TestRuntimeWritesFramesToStreamSink(t *testing.T) {
+	sink := audio.NewStreamSource(128, 44100)
+	src := newFakeSource()
+	rt := NewRuntime(sink, src)
+	if err := rt.Start(context.Background()); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	defer rt.Stop()
+
+	src.chunks <- PCMChunk{
+		Channels:     2,
+		SampleRateHz: 44100,
+		Samples:      []int32{1000 << 16, -1000 << 16},
+	}
+
+	deadline := time.Now().Add(1 * time.Second)
+	for sink.Available() < 1 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if sink.Available() < 1 {
+		t.Fatal("expected at least one frame in sink")
+	}
+}