From 7f65004cd435763227b71a328d21707f5181b26b Mon Sep 17 00:00:00 2001 From: w1ndyb0y Date: Fri, 10 Jul 2026 16:05:19 +0800 Subject: [PATCH] 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 | 2 +- app/app_test.go | 182 ++++++++++++++++++++++++++++++++ config/loader_test.go | 200 ++++++++++++++++++++++++++++++++++++ serial/serial_test.go | 100 ++++++++++++++++++ storage/store_test.go | 97 +++++++++++++++++ telegram/parser_test.go | 100 ++++++++++++++++++ transport/transport_test.go | 173 ++++++++++++++++++++++++++++++- 7 files changed, 852 insertions(+), 2 deletions(-) create mode 100644 app/app_test.go create mode 100644 config/loader_test.go create mode 100644 serial/serial_test.go diff --git a/app/app.go b/app/app.go index 5c6b21c..06891d5 100644 --- a/app/app.go +++ b/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 diff --git a/app/app_test.go b/app/app_test.go new file mode 100644 index 0000000..c62e7ca --- /dev/null +++ b/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) + } +} \ No newline at end of file diff --git a/config/loader_test.go b/config/loader_test.go new file mode 100644 index 0000000..eb0b4db --- /dev/null +++ b/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) + } +} \ No newline at end of file diff --git a/serial/serial_test.go b/serial/serial_test.go new file mode 100644 index 0000000..8607d58 --- /dev/null +++ b/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) +} \ No newline at end of file diff --git a/storage/store_test.go b/storage/store_test.go index 3d9bc83..f496251 100644 --- a/storage/store_test.go +++ b/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) + } +} \ No newline at end of file diff --git a/telegram/parser_test.go b/telegram/parser_test.go index 1ea988a..35f9009 100644 --- a/telegram/parser_test.go +++ b/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) + } +} \ No newline at end of file diff --git a/transport/transport_test.go b/transport/transport_test.go index 567bd06..1e54305 100644 --- a/transport/transport_test.go +++ b/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) + } +} \ No newline at end of file