ingest: add integration smoke coverage

cmd/fmrtx/main.go

@@ -7,6 +7,7 @@ import (
 	"log"
 	"os"
 	"os/signal"
+	"strings"
 	"syscall"
 	"time"
 
@@ -176,7 +177,7 @@ func runTXMode(cfg cfgpkg.Config, driver platform.SoapyDriver, autoStart bool, a
 	var streamSrc *audio.StreamSource
 	var ingestRuntime *ingest.Runtime
 	var ingress ctrlpkg.AudioIngress
-	if cfg.Ingest.Kind != "" && cfg.Ingest.Kind != "none" {
+	if ingestEnabled(cfg.Ingest.Kind) {
 		rate := ingestfactory.SampleRateForKind(cfg)
 		bufferFrames := rate * 2
 		if bufferFrames <= 0 {
@@ -241,6 +242,11 @@ func runTXMode(cfg cfgpkg.Config, driver platform.SoapyDriver, autoStart bool, a
 	log.Println("shutdown complete")
 }
 
+func ingestEnabled(kind string) bool {
+	normalized := strings.ToLower(strings.TrimSpace(kind))
+	return normalized != "" && normalized != "none"
+}
+
 func applyLegacyAudioFlags(cfg cfgpkg.Config, audioStdin bool, audioRate int, audioHTTP bool) cfgpkg.Config {
 	if audioRate > 0 {
 		cfg.Ingest.Stdin.SampleRateHz = audioRate

cmd/fmrtx/main_test.go

@@ -9,6 +9,28 @@ import (
 	"github.com/jan/fm-rds-tx/internal/platform"
 )
 
+func TestIngestEnabled(t *testing.T) {
+	tests := []struct {
+		name string
+		kind string
+		want bool
+	}{
+		{name: "empty", kind: "", want: false},
+		{name: "none", kind: "none", want: false},
+		{name: "none uppercase and spaces", kind: "  NONE  ", want: false},
+		{name: "stdin", kind: "stdin", want: true},
+		{name: "http raw uppercase", kind: " HTTP-RAW ", want: true},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := ingestEnabled(tc.kind); got != tc.want {
+				t.Fatalf("ingestEnabled(%q)=%v want %v", tc.kind, got, tc.want)
+			}
+		})
+	}
+}
+
 func TestTxBridgeExportsQueueStats(t *testing.T) {
 	cfg := cfgpkg.Default()
 	driver := platform.NewSimulatedDriver(nil)

internal/ingest/factory/factory_test.go

@@ -34,6 +34,21 @@ func TestBuildSourceHTTPRawProvidesIngress(t *testing.T) {
 	}
 }
 
+func TestBuildSourceKindIsNormalized(t *testing.T) {
+	cfg := config.Default()
+	cfg.Ingest.Kind = "  HTTP-RAW "
+	src, ingress, err := BuildSource(cfg, Deps{})
+	if err != nil {
+		t.Fatalf("build source: %v", err)
+	}
+	if src == nil || ingress == nil {
+		t.Fatalf("expected source and ingress for normalized http-raw kind")
+	}
+	if got := src.Descriptor().Kind; got != "http-raw" {
+		t.Fatalf("source kind=%q want http-raw", got)
+	}
+}
+
 func TestBuildSourceStdin(t *testing.T) {
 	cfg := config.Default()
 	cfg.Ingest.Kind = "stdin"

internal/ingest/factory/ingest_smoke_test.go

@@ -0,0 +1,76 @@
+package factory
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/jan/fm-rds-tx/internal/audio"
+	"github.com/jan/fm-rds-tx/internal/config"
+	"github.com/jan/fm-rds-tx/internal/ingest"
+)
+
+func TestHTTPRawFactoryToRuntimeSmoke(t *testing.T) {
+	cfg := config.Default()
+	cfg.Ingest.Kind = "http-raw"
+	cfg.Ingest.HTTPRaw.SampleRateHz = 44100
+	cfg.Ingest.HTTPRaw.Channels = 2
+
+	src, ingress, err := BuildSource(cfg, Deps{})
+	if err != nil {
+		t.Fatalf("build source: %v", err)
+	}
+	if src == nil || ingress == nil {
+		t.Fatalf("expected source and ingress for kind=http-raw")
+	}
+
+	sink := audio.NewStreamSource(128, cfg.Ingest.HTTPRaw.SampleRateHz)
+	rt := ingest.NewRuntime(sink, src)
+	if err := rt.Start(context.Background()); err != nil {
+		t.Fatalf("runtime start: %v", err)
+	}
+	defer rt.Stop()
+
+	// Two stereo frames: L1,R1,L2,R2 (S16LE).
+	frames, err := ingress.WritePCM16([]byte{
+		0xE8, 0x03, 0x18, 0xFC,
+		0xD0, 0x07, 0x30, 0xF8,
+	})
+	if err != nil {
+		t.Fatalf("write pcm16: %v", err)
+	}
+	if frames != 2 {
+		t.Fatalf("frames=%d want 2", frames)
+	}
+
+	waitForSinkFrames(t, sink, 2)
+
+	stats := rt.Stats()
+	if stats.Active.Kind != "http-raw" {
+		t.Fatalf("active kind=%q want http-raw", stats.Active.Kind)
+	}
+	if stats.Source.ChunksIn != 1 {
+		t.Fatalf("source chunksIn=%d want 1", stats.Source.ChunksIn)
+	}
+	if stats.Source.SamplesIn != 4 {
+		t.Fatalf("source samplesIn=%d want 4", stats.Source.SamplesIn)
+	}
+	if stats.Runtime.State != "running" {
+		t.Fatalf("runtime state=%q want running", stats.Runtime.State)
+	}
+	if stats.Runtime.LastChunkAt.IsZero() {
+		t.Fatalf("runtime lastChunkAt should be set")
+	}
+}
+
+func waitForSinkFrames(t *testing.T, sink *audio.StreamSource, minFrames int) {
+	t.Helper()
+	deadline := time.Now().Add(1 * time.Second)
+	for time.Now().Before(deadline) {
+		if sink.Available() >= minFrames {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("timeout waiting for sink frames: have=%d want>=%d", sink.Available(), minFrames)
+}