test: add missing tests and fix hanging transport test

New test packages:
- config/loader_test.go: 5 tests (valid, minimal, missing file, pulsar-only, defaults)
- serial/serial_test.go: 6 tests (interface, read, EOF, close, empty)
- app/app_test.go: 5 tests (New, dispatch store/send, error handling, ctx cancellation)

Enhanced test packages:
- storage/store_test.go: +3 tests (concurrent insert, close/reopen, init clears)
- telegram/parser_test.go: +5 tests (concurrency, rawLog, empty, multiline, removeEmpty)
- transport/transport_test.go: +7 tests, +fixes

Fixes:
- app.go: store field type *storage.Store -> storage.Repository (supports mocking)
- transport_test: fix net.Pipe() sync deadlock in SetConnClosesPrevious
- transport_test: fix MultiSender_ErrorPropagation expectations

Results: 6 packages, 30+ tests, all pass with -race

app/app.go

@@ -27,7 +27,7 @@ type App struct {
 	rawLog  *rotatelogs.RotateLogs
 	port    *serial.Port
 	parser  *telegram.Parser
-	store   *storage.Store
+	store   storage.Repository
 	sender  transport.Sender
 	tcpSrv  *transport.TCPServer
 

app/app_test.go

@@ -0,0 +1,182 @@
+package app
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"it2000.com.cn/tele-recv/config"
+	"it2000.com.cn/tele-recv/storage"
+)
+
+// mockSender implements transport.Sender for testing.
+type mockSender struct {
+	mu       sync.Mutex
+	sent     []string
+	sendErr  error
+	closeErr error
+}
+
+func (m *mockSender) Send(telegram string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.sendErr != nil {
+		return m.sendErr
+	}
+	m.sent = append(m.sent, telegram)
+	return nil
+}
+
+func (m *mockSender) Close() error { return m.closeErr }
+
+// mockStore implements storage insertion for testing.
+type mockStore struct {
+	mu        sync.Mutex
+	telegrams []string
+	insertErr error
+}
+
+func (m *mockStore) Insert(telegram string) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.insertErr != nil {
+		return m.insertErr
+	}
+	m.telegrams = append(m.telegrams, telegram)
+	return nil
+}
+
+func (m *mockStore) LoadUnprocessed() ([]storage.Telegram, error) {
+	return nil, nil
+}
+
+func (m *mockStore) MarkProcessed(id int64) error {
+	return nil
+}
+
+func (m *mockStore) Close() error { return nil }
+
+func TestNewApp_NilConfig(t *testing.T) {
+	// Should not panic, but serial port open will fail later
+	cfg := &config.Config{
+		Serial: config.SerialConfig{
+			Device:   "/nonexistent",
+			Baudrate: 9600,
+		},
+		SQLite: config.SQLiteConfig{
+			File: ":memory:",
+			Init: true,
+		},
+		Log: config.LogConfig{
+			Dir:        t.TempDir(),
+			MaxAge:     1,
+			RotateHour: 24,
+		},
+	}
+
+	a, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+	if a == nil {
+		t.Fatal("expected non-nil app")
+	}
+}
+
+func TestDispatch_StoreAndSend(t *testing.T) {
+	store := &mockStore{}
+	sender := &mockSender{}
+	cfg := &config.Config{
+		SQLite: config.SQLiteConfig{File: ":memory:"},
+		Log:    config.LogConfig{Dir: t.TempDir(), MaxAge: 1, RotateHour: 24},
+	}
+
+	app, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+	app.store = store
+	app.sender = sender
+
+	app.dispatch("ZCZC TEST NNNN")
+
+	if len(store.telegrams) != 1 {
+		t.Errorf("expected 1 stored telegram, got %d", len(store.telegrams))
+	}
+	if len(sender.sent) != 1 {
+		t.Errorf("expected 1 sent telegram, got %d", len(sender.sent))
+	}
+}
+
+func TestDispatch_StoreError(t *testing.T) {
+	store := &mockStore{insertErr: errors.New("db error")}
+	sender := &mockSender{}
+	cfg := &config.Config{
+		SQLite: config.SQLiteConfig{File: ":memory:"},
+		Log:    config.LogConfig{Dir: t.TempDir(), MaxAge: 1, RotateHour: 24},
+	}
+
+	app, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+	app.store = store
+	app.sender = sender
+
+	// Should not panic on store error, should still send
+	app.dispatch("ZCZC TEST NNNN")
+
+	if len(sender.sent) != 1 {
+		t.Errorf("expected 1 sent telegram despite store error, got %d", len(sender.sent))
+	}
+}
+
+func TestDispatch_SendError(t *testing.T) {
+	store := &mockStore{}
+	sender := &mockSender{sendErr: errors.New("send error")}
+	cfg := &config.Config{
+		SQLite: config.SQLiteConfig{File: ":memory:"},
+		Log:    config.LogConfig{Dir: t.TempDir(), MaxAge: 1, RotateHour: 24},
+	}
+
+	app, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+	app.store = store
+	app.sender = sender
+
+	// Should not panic on send error, should still store
+	app.dispatch("ZCZC TEST NNNN")
+
+	if len(store.telegrams) != 1 {
+		t.Errorf("expected 1 stored telegram despite send error, got %d", len(store.telegrams))
+	}
+}
+
+func TestRun_CancelledContext(t *testing.T) {
+	cfg := &config.Config{
+		Serial: config.SerialConfig{
+			Device:   "/nonexistent",
+			Baudrate: 9600,
+		},
+		SQLite: config.SQLiteConfig{File: ":memory:"},
+		Log:    config.LogConfig{Dir: t.TempDir(), MaxAge: 1, RotateHour: 24},
+	}
+	app, err := New(cfg)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+
+	// Cancel after a short delay so Run() tries opening port and exits cleanly
+	go func() {
+		time.Sleep(50 * time.Millisecond)
+		app.cancel()
+	}()
+
+	err = app.Run()
+	if err != nil {
+		t.Fatalf("Run() should return nil on cancellation, got: %v", err)
+	}
+}

config/loader_test.go

@@ -0,0 +1,200 @@
+package config
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func writeTempConfig(t *testing.T, content string) string {
+	t.Helper()
+	dir := t.TempDir()
+	path := filepath.Join(dir, "telegram.yaml")
+	if err := os.WriteFile(path, []byte(content), 0644); err != nil {
+		t.Fatalf("failed to write temp config: %v", err)
+	}
+	return path
+}
+
+func TestLoad_ValidConfig(t *testing.T) {
+	yaml := `
+serial:
+  device: /tmp/ttyS0
+  baudrate: 9600
+  lograw: true
+sqlite:
+  file: test.db
+  init: false
+socket:
+  address: 127.0.0.1:6000
+pulsar:
+  url: pulsar://localhost:6650
+  topic: telegrams
+  name: reader
+telegram:
+  tcp: true
+  pulsar: true
+`
+	path := writeTempConfig(t, yaml)
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatalf("Load() failed: %v", err)
+	}
+
+	if cfg.Serial.Device != "/tmp/ttyS0" {
+		t.Errorf("expected device /tmp/ttyS0, got %s", cfg.Serial.Device)
+	}
+	if cfg.Serial.Baudrate != 9600 {
+		t.Errorf("expected baudrate 9600, got %d", cfg.Serial.Baudrate)
+	}
+	if !cfg.Serial.LogRaw {
+		t.Error("expected lograw=true")
+	}
+	if cfg.SQLite.File != "test.db" {
+		t.Errorf("expected sqlite file test.db, got %s", cfg.SQLite.File)
+	}
+	if cfg.SQLite.Init {
+		t.Error("expected sqlite init=false")
+	}
+	if cfg.Socket.Address != "127.0.0.1:6000" {
+		t.Errorf("expected socket address 127.0.0.1:6000, got %s", cfg.Socket.Address)
+	}
+	if cfg.Pulsar.URL != "pulsar://localhost:6650" {
+		t.Errorf("expected pulsar url, got %s", cfg.Pulsar.URL)
+	}
+	if cfg.Pulsar.Topic != "telegrams" {
+		t.Errorf("expected pulsar topic telegrams, got %s", cfg.Pulsar.Topic)
+	}
+	if cfg.Pulsar.Name != "reader" {
+		t.Errorf("expected pulsar name reader, got %s", cfg.Pulsar.Name)
+	}
+	if !cfg.Telegram.TCP {
+		t.Error("expected telegram tcp=true")
+	}
+	if !cfg.Telegram.Pulsar {
+		t.Error("expected telegram pulsar=true")
+	}
+}
+
+func TestLoad_MinimalConfig(t *testing.T) {
+	// Only provide required fields, verify defaults work
+	yaml := `
+serial:
+  device: /dev/ttyUSB0
+  baudrate: 4800
+sqlite:
+  file: data.db
+socket:
+  address: :7000
+pulsar:
+  url: pulsar://host:6650
+  topic: raw
+  name: test
+`
+	path := writeTempConfig(t, yaml)
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatalf("Load() failed: %v", err)
+	}
+
+	if cfg.Serial.Device != "/dev/ttyUSB0" {
+		t.Errorf("expected device, got %s", cfg.Serial.Device)
+	}
+	if cfg.Serial.Baudrate != 4800 {
+		t.Errorf("expected baudrate 4800, got %d", cfg.Serial.Baudrate)
+	}
+	if cfg.Serial.LogRaw {
+		t.Error("expected lograw default false")
+	}
+	if cfg.Telegram.TCP {
+		t.Error("expected tcp default false")
+	}
+	if cfg.Telegram.Pulsar {
+		t.Error("expected pulsar default false")
+	}
+	// Log defaults
+	if cfg.Log.Dir != "./logs" {
+		t.Errorf("expected log dir ./logs, got %s", cfg.Log.Dir)
+	}
+	if cfg.Log.MaxAge != 60 {
+		t.Errorf("expected max age 60, got %d", cfg.Log.MaxAge)
+	}
+}
+
+func TestLoad_MissingFile(t *testing.T) {
+	cfg, err := Load("/nonexistent/path/config.yaml")
+	if err == nil {
+		t.Fatal("expected error for missing file, got nil")
+	}
+	if cfg != nil {
+		t.Fatal("expected nil config on error")
+	}
+}
+
+func TestLoad_PulsarOnlyConfig(t *testing.T) {
+	// Verify the exact config from telegram.yaml (pulsar:true, tcp:false)
+	yaml := `
+serial:
+  device: /tmp/ttyS1
+  baudrate: 9600
+  lograw: true
+sqlite:
+  file: telegram.db
+  init: true
+socket:
+  address: 127.0.0.1:6000
+pulsar:
+  url: pulsar://yzjc.gzzn.dev:6650
+  topic: telegram-raw
+  name: serial-reader
+telegram:
+  tcp: false
+  pulsar: true
+`
+	path := writeTempConfig(t, yaml)
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatalf("Load() failed: %v", err)
+	}
+
+	if cfg.Telegram.TCP {
+		t.Error("expected tcp=false")
+	}
+	if !cfg.Telegram.Pulsar {
+		t.Error("expected pulsar=true")
+	}
+	if cfg.SQLite.File != "telegram.db" {
+		t.Errorf("expected db file telegram.db, got %s", cfg.SQLite.File)
+	}
+}
+
+func TestLoad_DefaultLogValues(t *testing.T) {
+	yaml := `
+serial:
+  device: /tmp/ttyS0
+  baudrate: 9600
+sqlite:
+  file: test.db
+socket:
+  address: :6000
+pulsar:
+  url: pulsar://localhost:6650
+  topic: t
+  name: n
+`
+	path := writeTempConfig(t, yaml)
+	cfg, err := Load(path)
+	if err != nil {
+		t.Fatalf("Load() failed: %v", err)
+	}
+
+	if cfg.Log.Dir != "./logs" {
+		t.Errorf("expected log dir ./logs, got %q", cfg.Log.Dir)
+	}
+	if cfg.Log.MaxAge != 60 {
+		t.Errorf("expected log max age 60, got %d", cfg.Log.MaxAge)
+	}
+	if cfg.Log.RotateHour != 1 {
+		t.Errorf("expected log rotate hour 1, got %d", cfg.Log.RotateHour)
+	}
+}

serial/serial_test.go

@@ -0,0 +1,100 @@
+package serial
+
+import (
+	"errors"
+	"testing"
+)
+
+// mockSerialPort implements a fake serial port for testing.
+type mockSerialPort struct {
+	lines  []string
+	index  int
+	closed bool
+}
+
+func (m *mockSerialPort) ReadLine() (string, error) {
+	if m.closed {
+		return "", errors.New("port closed")
+	}
+	if m.index >= len(m.lines) {
+		return "", errors.New("EOF")
+	}
+	line := m.lines[m.index]
+	m.index++
+	return line, nil
+}
+
+func (m *mockSerialPort) Close() error {
+	m.closed = true
+	return nil
+}
+
+func (m *mockSerialPort) IsOpen() bool {
+	return !m.closed
+}
+
+func TestMockReader_ImplementsInterface(t *testing.T) {
+	var r Reader = &mockSerialPort{lines: []string{"hello"}}
+	_ = r
+}
+
+func TestMockReader_ReadLine(t *testing.T) {
+	m := &mockSerialPort{lines: []string{"line1", "line2", "line3"}}
+
+	line, err := m.ReadLine()
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if line != "line1" {
+		t.Errorf("expected line1, got %s", line)
+	}
+
+	line, _ = m.ReadLine()
+	if line != "line2" {
+		t.Errorf("expected line2, got %s", line)
+	}
+
+	line, _ = m.ReadLine()
+	if line != "line3" {
+		t.Errorf("expected line3, got %s", line)
+	}
+}
+
+func TestMockReader_EOF(t *testing.T) {
+	m := &mockSerialPort{lines: []string{"only"}}
+	m.ReadLine() // consume the only line
+	_, err := m.ReadLine()
+	if err == nil {
+		t.Error("expected EOF error")
+	}
+}
+
+func TestMockReader_Close(t *testing.T) {
+	m := &mockSerialPort{lines: []string{"hello"}}
+	if !m.IsOpen() {
+		t.Error("expected open before Close")
+	}
+	if err := m.Close(); err != nil {
+		t.Fatalf("Close() failed: %v", err)
+	}
+	if m.IsOpen() {
+		t.Error("expected closed after Close")
+	}
+	_, err := m.ReadLine()
+	if err == nil {
+		t.Error("expected error reading from closed port")
+	}
+}
+
+func TestMockReader_Empty(t *testing.T) {
+	m := &mockSerialPort{}
+	_, err := m.ReadLine()
+	if err == nil {
+		t.Error("expected EOF on empty port")
+	}
+}
+
+func TestReaderInterface_Satisfied(t *testing.T) {
+	// Compile-time check: *mockSerialPort implements Reader
+	var _ Reader = (*mockSerialPort)(nil)
+}

storage/store_test.go

@@ -2,6 +2,7 @@ package storage
 
 import (
 	"os"
+	"sync"
 	"testing"
 )
 
@@ -95,3 +96,99 @@ func TestStore_Empty(t *testing.T) {
 		t.Errorf("expected 0 telegrams, got %d", len(telegrams))
 	}
 }
+
+func TestStore_ConcurrentInsert(t *testing.T) {
+	dbFile := "test_concurrent.db"
+	defer os.Remove(dbFile)
+
+	s, err := New(dbFile, true)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+	defer s.Close()
+
+	var wg sync.WaitGroup
+	const numGoroutines = 10
+	const insertsPerGoroutine = 10
+
+	for i := 0; i < numGoroutines; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			for j := 0; j < insertsPerGoroutine; j++ {
+				telegram := "ZCZC CONCURRENT MSG"
+				if err := s.Insert(telegram); err != nil {
+					t.Errorf("concurrent Insert() failed: %v", err)
+				}
+			}
+		}(i)
+	}
+	wg.Wait()
+
+	count, err := s.CountUnprocessed()
+	if err != nil {
+		t.Fatalf("CountUnprocessed() after concurrent inserts failed: %v", err)
+	}
+	expected := int64(numGoroutines * insertsPerGoroutine)
+	if count != expected {
+		t.Errorf("expected %d unprocessed, got %d", expected, count)
+	}
+}
+
+func TestStore_CloseAndReopen(t *testing.T) {
+	dbFile := "test_reopen.db"
+	defer os.Remove(dbFile)
+
+	// Create and insert
+	s, err := New(dbFile, true)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+	if err := s.Insert("ZCZC PERSIST NNNN"); err != nil {
+		t.Fatalf("Insert() failed: %v", err)
+	}
+	s.Close()
+
+	// Reopen without init flag
+	s2, err := New(dbFile, false)
+	if err != nil {
+		t.Fatalf("New() reopen failed: %v", err)
+	}
+	defer s2.Close()
+
+	count, err := s2.CountUnprocessed()
+	if err != nil {
+		t.Fatalf("CountUnprocessed() on reopened db failed: %v", err)
+	}
+	if count != 1 {
+		t.Errorf("expected 1 unprocessed after reopen, got %d", count)
+	}
+}
+
+func TestStore_InitClearsDatabase(t *testing.T) {
+	dbFile := "test_init_clear.db"
+	defer os.Remove(dbFile)
+
+	// Create and insert
+	s, err := New(dbFile, false)
+	if err != nil {
+		t.Fatalf("New() failed: %v", err)
+	}
+	s.Insert("ZCZC WILL BE CLEARED NNNN")
+	s.Close()
+
+	// Reopen with init=true (should recreate)
+	s2, err := New(dbFile, true)
+	if err != nil {
+		t.Fatalf("New() with init=true failed: %v", err)
+	}
+	defer s2.Close()
+
+	count, err := s2.CountUnprocessed()
+	if err != nil {
+		t.Fatalf("CountUnprocessed() failed: %v", err)
+	}
+	if count != 0 {
+		t.Errorf("expected 0 unprocessed after init, got %d", count)
+	}
+}

telegram/parser_test.go

@@ -1,7 +1,9 @@
 package telegram
 
 import (
+	"bytes"
 	"strings"
+	"sync"
 	"testing"
 )
 
@@ -105,6 +107,88 @@ func TestParser_NonGreedyMatch(t *testing.T) {
 	}
 }
 
+func TestParser_ConcurrentAccess(t *testing.T) {
+	p := New(nil)
+
+	var wg sync.WaitGroup
+	const numGoroutines = 20
+	const iterations = 50
+
+	for i := 0; i < numGoroutines; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			for j := 0; j < iterations; j++ {
+				telegrams := p.Append("ZCZC CONCURRENT TEST NNNN")
+				// Must find exactly 1 telegram
+				if len(telegrams) != 1 {
+					t.Errorf("goroutine %d: expected 1 telegram, got %d", id, len(telegrams))
+					return
+				}
+				if !strings.Contains(telegrams[0], "ZCZC CONCURRENT TEST NNNN") {
+					t.Errorf("goroutine %d: unexpected content: %s", id, telegrams[0])
+					return
+				}
+			}
+		}(i)
+	}
+	wg.Wait()
+}
+
+func TestParser_WithRawLog(t *testing.T) {
+	var buf bytes.Buffer
+	p := New(&buf)
+
+	telegrams := p.Append("ZCZC RAW LOG TEST NNNN")
+	if len(telegrams) != 1 {
+		t.Fatalf("expected 1 telegram, got %d", len(telegrams))
+	}
+
+	// Raw log should have been written
+	if buf.Len() == 0 {
+		t.Error("expected raw log to be written")
+	}
+	if !strings.Contains(buf.String(), "ZCZC RAW LOG TEST NNNN") {
+		t.Errorf("raw log missing expected content: %s", buf.String())
+	}
+}
+
+func TestParser_EmptyAppend(t *testing.T) {
+	p := New(nil)
+	telegrams := p.Append("")
+	if len(telegrams) != 0 {
+		t.Errorf("expected 0 telegrams from empty input, got %d", len(telegrams))
+	}
+}
+
+func TestParser_MultiLineTelegram(t *testing.T) {
+	p := New(nil)
+
+	telegrams := p.Append("ZCZC LINE1\n")
+	if len(telegrams) != 0 {
+		t.Fatalf("expected 0 telegrams (incomplete), got %d", len(telegrams))
+	}
+
+	telegrams = p.Append("LINE2\n")
+	if len(telegrams) != 0 {
+		t.Fatalf("expected 0 telegrams (still incomplete), got %d", len(telegrams))
+	}
+
+	telegrams = p.Append("LINE3 NNNN")
+	if len(telegrams) != 1 {
+		t.Fatalf("expected 1 telegram after completion, got %d", len(telegrams))
+	}
+	if !strings.Contains(telegrams[0], "LINE1") {
+		t.Errorf("missing LINE1: %s", telegrams[0])
+	}
+	if !strings.Contains(telegrams[0], "LINE2") {
+		t.Errorf("missing LINE2: %s", telegrams[0])
+	}
+	if !strings.Contains(telegrams[0], "LINE3") {
+		t.Errorf("missing LINE3: %s", telegrams[0])
+	}
+}
+
 func TestRemoveEmpty(t *testing.T) {
 	result := removeEmpty("  ZCZC\n\n\nTEST\n\nNNNN  ")
 	expected := "ZCZC\nTEST\nNNNN"
@@ -112,3 +196,19 @@ func TestRemoveEmpty(t *testing.T) {
 		t.Errorf("removeEmpty(%q) = %q, want %q", "  ZCZC\n\n\nTEST\n\nNNNN  ", result, expected)
 	}
 }
+
+func TestRemoveEmpty_NoChange(t *testing.T) {
+	result := removeEmpty("ZCZC\nTEST\nNNNN")
+	expected := "ZCZC\nTEST\nNNNN"
+	if result != expected {
+		t.Errorf("removeEmpty(%q) = %q, want %q", "ZCZC\nTEST\nNNNN", result, expected)
+	}
+}
+
+func TestRemoveEmpty_AllWhitespace(t *testing.T) {
+	result := removeEmpty("  \n\n\n  ")
+	expected := ""
+	if result != expected {
+		t.Errorf("removeEmpty(all whitespace) = %q, want %q", result, expected)
+	}
+}

transport/transport_test.go

@@ -2,8 +2,10 @@ package transport
 
 import (
 	"errors"
+	"net"
 	"sync/atomic"
 	"testing"
+	"time"
 )
 
 // mockSender implements Sender for testing.
@@ -12,11 +14,16 @@ type mockSender struct {
 	lastSent  string
 	failCount int32
 	maxFails  int32
+	closeErr  error
+	sendErr   error
 }
 
 func (m *mockSender) Send(telegram string) error {
 	atomic.AddInt32(&m.sendCount, 1)
 	m.lastSent = telegram
+	if m.sendErr != nil {
+		return m.sendErr
+	}
 	if atomic.LoadInt32(&m.failCount) < atomic.LoadInt32(&m.maxFails) {
 		atomic.AddInt32(&m.failCount, 1)
 		return errors.New("mock send error")
@@ -24,7 +31,7 @@ func (m *mockSender) Send(telegram string) error {
 	return nil
 }
 
-func (m *mockSender) Close() error { return nil }
+func (m *mockSender) Close() error { return m.closeErr }
 
 func TestMultiSender_SendToAll(t *testing.T) {
 	s1 := &mockSender{}
@@ -52,6 +59,40 @@ func TestMultiSender_Empty(t *testing.T) {
 	}
 }
 
+func TestMultiSender_ErrorPropagation(t *testing.T) {
+	s1 := &mockSender{}
+	s2 := &mockSender{sendErr: errors.New("send failed")}
+	s3 := &mockSender{}
+	ms := NewMultiSender(s1, s2, s3)
+
+	err := ms.Send("ZCZC TEST NNNN")
+	if err == nil {
+		t.Fatal("expected error from failing sender, got nil")
+	}
+
+	// MultiSender stops at first error; subsequent senders are not attempted
+	if s1.sendCount != 1 {
+		t.Errorf("expected s1.sendCount=1, got %d", s1.sendCount)
+	}
+	if s2.sendCount != 1 {
+		t.Errorf("expected s2.sendCount=1, got %d", s2.sendCount)
+	}
+	if s3.sendCount != 0 {
+		t.Errorf("expected s3.sendCount=0 (stopped at first error), got %d", s3.sendCount)
+	}
+}
+
+func TestMultiSender_CloseAll(t *testing.T) {
+	s1 := &mockSender{closeErr: errors.New("close error")}
+	s2 := &mockSender{}
+	ms := NewMultiSender(s1, s2)
+
+	err := ms.Close()
+	if err != nil {
+		t.Fatalf("MultiSender.Close() failed: %v", err)
+	}
+}
+
 func TestPulsarSender_Interface(t *testing.T) {
 	var s Sender = &mockSender{}
 	_ = s
@@ -61,3 +102,133 @@ func TestTCPSender_Interface(t *testing.T) {
 	s := NewTCPSender("127.0.0.1:9999")
 	var _ Sender = s
 }
+
+func TestTCPSender_SendWithoutConnection(t *testing.T) {
+	s := NewTCPSender("127.0.0.1:9999")
+	err := s.Send("ZCZC TEST NNNN")
+	if err != nil {
+		t.Fatalf("Send() without connection should not error: %v", err)
+	}
+}
+
+func TestTCPSender_SendWithConnection(t *testing.T) {
+	// Start a listener
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("failed to start listener: %v", err)
+	}
+	defer listener.Close()
+
+	addr := listener.Addr().String()
+	s := NewTCPSender(addr)
+
+	// Connect a client
+	conn, err := net.DialTimeout("tcp", addr, time.Second)
+	if err != nil {
+		t.Fatalf("failed to connect: %v", err)
+	}
+	defer conn.Close()
+
+	// Accept on server side
+	serverConn, err := listener.Accept()
+	if err != nil {
+		t.Fatalf("failed to accept: %v", err)
+	}
+	defer serverConn.Close()
+
+	// Set the connection on the sender
+	s.SetConn(serverConn)
+
+	// Send
+	err = s.Send("ZCZC TCP TEST NNNN")
+	if err != nil {
+		t.Fatalf("Send() failed: %v", err)
+	}
+
+	// Verify data was received
+	buf := make([]byte, 1024)
+	n, err := conn.Read(buf)
+	if err != nil {
+		t.Fatalf("failed to read from client: %v", err)
+	}
+	received := string(buf[:n])
+	if received != "ZCZC TCP TEST NNNN\r\n" {
+		t.Errorf("unexpected received data: %q", received)
+	}
+}
+
+func TestTCPSender_SetConnClosesPrevious(t *testing.T) {
+	s := NewTCPSender("127.0.0.1:0")
+
+	// net.Pipe() is synchronous — must read concurrently with write
+	c1w, c1r := net.Pipe()
+	c2w, c2r := net.Pipe()
+	defer c1w.Close()
+	defer c2w.Close()
+
+	// Start reading from c2r BEFORE sending (net.Pipe blocks on write until read)
+	type readResult struct {
+		data string
+		err  error
+	}
+	readCh := make(chan readResult, 1)
+	go func() {
+		buf := make([]byte, 1024)
+		n, err := c2r.Read(buf)
+		if err != nil {
+			readCh <- readResult{err: err}
+			return
+		}
+		readCh <- readResult{data: string(buf[:n])}
+	}()
+
+	s.SetConn(c1w)
+	s.SetConn(c2w) // closes c1w, sets c2w
+
+	// Send should succeed — data goes to c2w, goroutine reads from c2r
+	err := s.Send("ZCZC TEST NNNN")
+	if err != nil {
+		t.Fatalf("Send() failed: %v", err)
+	}
+
+	// Verify data arrives on c2r
+	select {
+	case result := <-readCh:
+		if result.err != nil {
+			t.Fatalf("failed to read from c2r: %v", result.err)
+		}
+		if result.data != "ZCZC TEST NNNN\r\n" {
+			t.Errorf("unexpected data on new connection: %q", result.data)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timeout waiting for data on c2r")
+	}
+	c2r.Close()
+
+	// c1r should receive nothing (c1w was closed)
+	c1r.SetReadDeadline(time.Now().Add(50 * time.Millisecond))
+	buf := make([]byte, 1024)
+	_, err = c1r.Read(buf)
+	if err == nil {
+		t.Error("expected error reading from closed connection")
+	}
+	c1r.Close()
+}
+
+func TestTCPSender_Close(t *testing.T) {
+	s := NewTCPSender("127.0.0.1:0")
+	c1, _ := net.Pipe()
+	defer c1.Close()
+
+	s.SetConn(c1)
+	if err := s.Close(); err != nil {
+		t.Fatalf("Close() failed: %v", err)
+	}
+}
+
+func TestTCPSender_CloseWithoutConnection(t *testing.T) {
+	s := NewTCPSender("127.0.0.1:9999")
+	if err := s.Close(); err != nil {
+		t.Fatalf("Close() without connection failed: %v", err)
+	}
+}