package tailer_test import ( "os" "path/filepath" "testing" "workshop.perforce.com/p4lf/internal/config" "workshop.perforce.com/p4lf/internal/tailer" ) type nopLogger struct{} func (nopLogger) Infof(string, ...interface{}) {} func (nopLogger) Warnf(string, ...interface{}) {} func (nopLogger) Debugf(string, ...interface{}) {} func (nopLogger) Errorf(string, ...interface{}) {} func testConfig(t *testing.T, logPath string) config.Config { t.Helper() cfg := config.Defaults() cfg.P4LogFile = logPath cfg.StateFile = logPath + ".state" cfg.LogChunksDir = filepath.Dir(logPath) cfg.ReadFromStart = true cfg.MaxRotationRecoverySize = 500 * 1024 * 1024 return cfg } func appendToFile(t *testing.T, path string, data []byte) { t.Helper() f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0644) if err != nil { t.Fatalf("appendToFile open %q: %v", path, err) } defer f.Close() if _, err := f.Write(data); err != nil { t.Fatalf("appendToFile write %q: %v", path, err) } } // TestFlush_ReadFromStart verifies that on first run (no state file), all // existing content in the P4LOG is returned in the first flush. func TestFlush_ReadFromStart(t *testing.T) { dir := t.TempDir() logPath := filepath.Join(dir, "log") content := []byte("line1\nline2\n") if err := os.WriteFile(logPath, content, 0644); err != nil { t.Fatal(err) } cfg := testConfig(t, logPath) cfg.ReadFromStart = true tr, err := tailer.New(cfg, nopLogger{}) if err != nil { t.Fatalf("New: %v", err) } defer tr.Close() chunk, err := tr.Flush() if err != nil { t.Fatalf("Flush: %v", err) } if chunk == nil { t.Fatal("expected chunk, got nil") } if string(chunk.Data) != string(content) { t.Errorf("chunk data: got %q, want %q", chunk.Data, content) } if chunk.StartOffset != 0 { t.Errorf("StartOffset: got %d, want 0", chunk.StartOffset) } if chunk.EndOffset != int64(len(content)) { t.Errorf("EndOffset: got %d, want %d", chunk.EndOffset, len(content)) } } // TestFlush_ReadFromEnd verifies that with ReadFromStart=false, existing content // is skipped and only newly appended data is returned. func TestFlush_ReadFromEnd(t *testing.T) { dir := t.TempDir() logPath := filepath.Join(dir, "log") existing := []byte("existing content\n") if err := os.WriteFile(logPath, existing, 0644); err != nil { t.Fatal(err) } cfg := testConfig(t, logPath) cfg.ReadFromStart = false tr, err := tailer.New(cfg, nopLogger{}) if err != nil { t.Fatalf("New: %v", err) } defer tr.Close() // First flush: no new data (started at EOF). chunk, err := tr.Flush() if err != nil { t.Fatalf("Flush: %v", err) } if chunk != nil { t.Errorf("expected nil chunk (started at EOF), got %q", chunk.Data) } // Append and flush: should return only the appended bytes. appended := []byte("new stuff\n") appendToFile(t, logPath, appended) chunk, err = tr.Flush() if err != nil { t.Fatalf("Flush after append: %v", err) } if chunk == nil { t.Fatal("expected chunk after append, got nil") } if string(chunk.Data) != string(appended) { t.Errorf("appended chunk: got %q, want %q", chunk.Data, appended) } if chunk.StartOffset != int64(len(existing)) { t.Errorf("StartOffset: got %d, want %d", chunk.StartOffset, len(existing)) } } // TestFlush_StateResume verifies that after saving state and creating a new // tailer, the second tailer resumes exactly where the first left off with no // duplicate or missing bytes. func TestFlush_StateResume(t *testing.T) { dir := t.TempDir() logPath := filepath.Join(dir, "log") initial := []byte("initial data\n") if err := os.WriteFile(logPath, initial, 0644); err != nil { t.Fatal(err) } cfg := testConfig(t, logPath) // First tailer: read initial content, save state, close. tr1, err := tailer.New(cfg, nopLogger{}) if err != nil { t.Fatalf("New: %v", err) } chunk1, err := tr1.Flush() if err != nil || chunk1 == nil { t.Fatalf("Flush 1: err=%v chunk=%v", err, chunk1) } if err := tr1.SaveState(); err != nil { t.Fatalf("SaveState: %v", err) } tr1.Close() // Append more data before creating second tailer. more := []byte("more data\n") appendToFile(t, logPath, more) // Second tailer: must resume from saved offset, returning only new bytes. tr2, err := tailer.New(cfg, nopLogger{}) if err != nil { t.Fatalf("New (resume): %v", err) } defer tr2.Close() chunk2, err := tr2.Flush() if err != nil || chunk2 == nil { t.Fatalf("Flush (resume): err=%v chunk=%v", err, chunk2) } if string(chunk2.Data) != string(more) { t.Errorf("resumed chunk: got %q, want %q", chunk2.Data, more) } if chunk2.StartOffset != int64(len(initial)) { t.Errorf("StartOffset: got %d, want %d", chunk2.StartOffset, len(initial)) } if chunk2.EndOffset != int64(len(initial)+len(more)) { t.Errorf("EndOffset: got %d, want %d", chunk2.EndOffset, len(initial)+len(more)) } } // TestFlush_RotationDetected verifies that when the P4LOG is replaced (inode // changes), the tailer detects the rotation and reads from the start of the // new file. func TestFlush_RotationDetected(t *testing.T) { dir := t.TempDir() logPath := filepath.Join(dir, "log") original := []byte("original log data\n") if err := os.WriteFile(logPath, original, 0644); err != nil { t.Fatal(err) } cfg := testConfig(t, logPath) tr, err := tailer.New(cfg, nopLogger{}) if err != nil { t.Fatalf("New: %v", err) } defer tr.Close() // Flush original content fully. chunk, err := tr.Flush() if err != nil || chunk == nil { t.Fatalf("Flush (pre-rotation): err=%v chunk=%v", err, chunk) } if string(chunk.Data) != string(original) { t.Errorf("pre-rotation data: got %q, want %q", chunk.Data, original) } // Simulate rotation: remove old file, create new one at the same path. if err := os.Remove(logPath); err != nil { t.Fatal(err) } newContent := []byte("new log after rotation\n") if err := os.WriteFile(logPath, newContent, 0644); err != nil { t.Fatal(err) } // Flush after rotation: tailer detects inode change, reads from offset 0. chunk, err = tr.Flush() if err != nil { t.Fatalf("Flush (post-rotation): %v", err) } if chunk == nil { t.Fatal("expected chunk after rotation, got nil") } if string(chunk.Data) != string(newContent) { t.Errorf("post-rotation data: got %q, want %q", chunk.Data, newContent) } if chunk.StartOffset != 0 { t.Errorf("StartOffset after rotation: got %d, want 0", chunk.StartOffset) } if chunk.EndOffset != int64(len(newContent)) { t.Errorf("EndOffset after rotation: got %d, want %d", chunk.EndOffset, len(newContent)) } } // TestFlush_NoNewData verifies that Flush returns nil when there is no new data. func TestFlush_NoNewData(t *testing.T) { dir := t.TempDir() logPath := filepath.Join(dir, "log") if err := os.WriteFile(logPath, []byte{}, 0644); err != nil { t.Fatal(err) } cfg := testConfig(t, logPath) tr, err := tailer.New(cfg, nopLogger{}) if err != nil { t.Fatalf("New: %v", err) } defer tr.Close() chunk, err := tr.Flush() if err != nil { t.Fatalf("Flush: %v", err) } if chunk != nil { t.Errorf("expected nil from empty file, got %q", chunk.Data) } } // TestFlush_OffsetAccumulation verifies that consecutive flushes produce // contiguous, non-overlapping offset windows. func TestFlush_OffsetAccumulation(t *testing.T) { dir := t.TempDir() logPath := filepath.Join(dir, "log") if err := os.WriteFile(logPath, []byte{}, 0644); err != nil { t.Fatal(err) } cfg := testConfig(t, logPath) tr, err := tailer.New(cfg, nopLogger{}) if err != nil { t.Fatalf("New: %v", err) } defer tr.Close() batch1 := []byte("batch one\n") appendToFile(t, logPath, batch1) chunk1, err := tr.Flush() if err != nil || chunk1 == nil { t.Fatalf("Flush batch1: err=%v chunk=%v", err, chunk1) } batch2 := []byte("batch two\n") appendToFile(t, logPath, batch2) chunk2, err := tr.Flush() if err != nil || chunk2 == nil { t.Fatalf("Flush batch2: err=%v chunk=%v", err, chunk2) } if chunk2.StartOffset != chunk1.EndOffset { t.Errorf("gap or overlap: chunk1.End=%d chunk2.Start=%d", chunk1.EndOffset, chunk2.StartOffset) } if chunk2.EndOffset != chunk1.EndOffset+int64(len(batch2)) { t.Errorf("chunk2.EndOffset: got %d, want %d", chunk2.EndOffset, chunk1.EndOffset+int64(len(batch2))) } if string(chunk2.Data) != string(batch2) { t.Errorf("chunk2 data: got %q, want %q", chunk2.Data, batch2) } }