test: add comprehensive tests to improve coverage

- observability: added tests for LogLevel.String(), Warn(), Warnf(), Infof(),
  Debugf(), Error(), WithFields(), and concurrent access patterns
- file: added tests for Reopen(), Close(), empty/whitespace paths,
  validateFilePath allowed/rejected paths, concurrent writes, Flush(),
  and marshal errors
- config: added tests for TimeWindowConfig.GetDuration(),
  CorrelationConfig getters, validation scenarios (no inputs, no outputs,
  duplicate sockets, ClickHouse validation), and LogConfig.GetLevel()

Coverage improvements:
- observability: 57.7% → 79.5%
- file: 68.6% → 78.6%
- config: 69.8% → 97.7%
- total: 68.6% → 74.4%

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

internal/adapters/outbound/file/sink_test.go

@@ -216,3 +216,213 @@ func TestFileSink_InvalidPath(t *testing.T) {
 		t.Error("expected error for invalid path")
 	}
 }
+
+func TestFileSink_Reopen(t *testing.T) {
+	tmpDir := t.TempDir()
+	testPath := filepath.Join(tmpDir, "test.log")
+
+	sink, err := NewFileSink(Config{Path: testPath})
+	if err != nil {
+		t.Fatalf("failed to create sink: %v", err)
+	}
+
+	// Write initial data
+	log := domain.CorrelatedLog{SrcIP: "192.168.1.1", SrcPort: 8080}
+	if err := sink.Write(context.Background(), log); err != nil {
+		t.Fatalf("failed to write: %v", err)
+	}
+
+	// Reopen should close and reopen the file
+	err = sink.Reopen()
+	if err != nil {
+		t.Errorf("expected no error on Reopen, got %v", err)
+	}
+
+	// Write after reopen
+	log2 := domain.CorrelatedLog{SrcIP: "192.168.1.2", SrcPort: 8081}
+	if err := sink.Write(context.Background(), log2); err != nil {
+		t.Fatalf("failed to write after reopen: %v", err)
+	}
+
+	sink.Close()
+
+	// Verify both writes are present
+	data, err := os.ReadFile(testPath)
+	if err != nil {
+		t.Fatalf("failed to read file: %v", err)
+	}
+
+	lines := 0
+	for _, b := range data {
+		if b == '\n' {
+			lines++
+		}
+	}
+
+	if lines != 2 {
+		t.Errorf("expected 2 lines after reopen, got %d", lines)
+	}
+}
+
+func TestFileSink_Close(t *testing.T) {
+	tmpDir := t.TempDir()
+	testPath := filepath.Join(tmpDir, "test.log")
+
+	sink, err := NewFileSink(Config{Path: testPath})
+	if err != nil {
+		t.Fatalf("failed to create sink: %v", err)
+	}
+
+	// Close should succeed
+	err = sink.Close()
+	if err != nil {
+		t.Errorf("expected no error on Close, got %v", err)
+	}
+
+	// Write after close should fail or reopen
+	log := domain.CorrelatedLog{SrcIP: "192.168.1.1", SrcPort: 8080}
+	err = sink.Write(context.Background(), log)
+	if err != nil {
+		// Expected - file was closed
+		t.Logf("write after close returned error (expected): %v", err)
+	}
+}
+
+func TestFileSink_EmptyPath(t *testing.T) {
+	_, err := NewFileSink(Config{Path: ""})
+	if err == nil {
+		t.Error("expected error for empty path")
+	}
+}
+
+func TestFileSink_WhitespacePath(t *testing.T) {
+	_, err := NewFileSink(Config{Path: "   "})
+	if err == nil {
+		t.Error("expected error for whitespace-only path")
+	}
+}
+
+func TestFileSink_ValidateFilePath_AllowedRoots(t *testing.T) {
+	// Test paths under allowed roots
+	allowedPaths := []string{
+		"/var/log/logcorrelator/correlated.log",
+		"/var/log/test.log",
+		"/tmp/test.log",
+		"/tmp/subdir/test.log",
+		"relative/path/test.log",
+		"./test.log",
+	}
+
+	for _, path := range allowedPaths {
+		err := validateFilePath(path)
+		if err != nil {
+			t.Errorf("validateFilePath(%q) unexpected error: %v", path, err)
+		}
+	}
+}
+
+func TestFileSink_ValidateFilePath_RejectedPaths(t *testing.T) {
+	// Test paths that should be rejected
+	rejectedPaths := []string{
+		"",
+		"   ",
+		"/etc/passwd",
+		"/etc/logcorrelator/test.log",
+		"/root/test.log",
+		"/home/user/test.log",
+		"/var/logevil/test.log",
+	}
+
+	for _, path := range rejectedPaths {
+		err := validateFilePath(path)
+		if err == nil {
+			t.Errorf("validateFilePath(%q) should have been rejected", path)
+		}
+	}
+}
+
+func TestFileSink_ConcurrentWrites(t *testing.T) {
+	tmpDir := t.TempDir()
+	testPath := filepath.Join(tmpDir, "test.log")
+
+	sink, err := NewFileSink(Config{Path: testPath})
+	if err != nil {
+		t.Fatalf("failed to create sink: %v", err)
+	}
+	defer sink.Close()
+
+	done := make(chan bool)
+	for i := 0; i < 10; i++ {
+		go func(n int) {
+			log := domain.CorrelatedLog{SrcIP: "192.168.1.1", SrcPort: 8080 + n}
+			sink.Write(context.Background(), log)
+			done <- true
+		}(i)
+	}
+
+	for i := 0; i < 10; i++ {
+		<-done
+	}
+
+	// Verify all writes completed
+	data, err := os.ReadFile(testPath)
+	if err != nil {
+		t.Fatalf("failed to read file: %v", err)
+	}
+
+	lines := 0
+	for _, b := range data {
+		if b == '\n' {
+			lines++
+		}
+	}
+
+	if lines != 10 {
+		t.Errorf("expected 10 lines from concurrent writes, got %d", lines)
+	}
+}
+
+func TestFileSink_Flush(t *testing.T) {
+	tmpDir := t.TempDir()
+	testPath := filepath.Join(tmpDir, "test.log")
+
+	sink, err := NewFileSink(Config{Path: testPath})
+	if err != nil {
+		t.Fatalf("failed to create sink: %v", err)
+	}
+	defer sink.Close()
+
+	log := domain.CorrelatedLog{SrcIP: "192.168.1.1", SrcPort: 8080}
+	if err := sink.Write(context.Background(), log); err != nil {
+		t.Fatalf("failed to write: %v", err)
+	}
+
+	// Flush should succeed
+	err = sink.Flush(context.Background())
+	if err != nil {
+		t.Errorf("expected no error on Flush, got %v", err)
+	}
+}
+
+func TestFileSink_MarshalError(t *testing.T) {
+	tmpDir := t.TempDir()
+	testPath := filepath.Join(tmpDir, "test.log")
+
+	sink, err := NewFileSink(Config{Path: testPath})
+	if err != nil {
+		t.Fatalf("failed to create sink: %v", err)
+	}
+	defer sink.Close()
+
+	// Create a log with unmarshalable data (channel)
+	log := domain.CorrelatedLog{
+		SrcIP:   "192.168.1.1",
+		SrcPort: 8080,
+		Fields:  map[string]any{"chan": make(chan int)},
+	}
+
+	err = sink.Write(context.Background(), log)
+	if err == nil {
+		t.Error("expected error when marshaling unmarshalable data")
+	}
+}

internal/config/config_test.go

@@ -588,3 +588,353 @@ correlation:
 		t.Error("expected outputs.stdout.enabled to be true")
 	}
 }
+
+func TestTimeWindowConfig_GetDuration(t *testing.T) {
+	tests := []struct {
+		name     string
+		config   TimeWindowConfig
+		expected time.Duration
+	}{
+		{
+			name:     "seconds",
+			config:   TimeWindowConfig{Value: 5, Unit: "s"},
+			expected: 5 * time.Second,
+		},
+		{
+			name:     "milliseconds",
+			config:   TimeWindowConfig{Value: 500, Unit: "ms"},
+			expected: 500 * time.Millisecond,
+		},
+		{
+			name:     "default on zero value",
+			config:   TimeWindowConfig{Value: 0, Unit: "s"},
+			expected: 1 * time.Second,
+		},
+		{
+			name:     "unknown unit defaults to seconds",
+			config:   TimeWindowConfig{Value: 5, Unit: "unknown"},
+			expected: 5 * time.Second,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := tt.config.GetDuration()
+			if result != tt.expected {
+				t.Errorf("expected %v, got %v", tt.expected, result)
+			}
+		})
+	}
+}
+
+func TestCorrelationConfig_GetTimeWindow(t *testing.T) {
+	cfg := CorrelationConfig{
+		TimeWindow: TimeWindowConfig{Value: 2, Unit: "s"},
+	}
+
+	result := cfg.GetTimeWindow()
+	if result != 2*time.Second {
+		t.Errorf("expected 2s, got %v", result)
+	}
+}
+
+func TestCorrelationConfig_GetTimeWindow_Default(t *testing.T) {
+	cfg := CorrelationConfig{
+		TimeWindow: TimeWindowConfig{Value: 0, Unit: "s"},
+	}
+
+	result := cfg.GetTimeWindow()
+	if result != 1*time.Second {
+		t.Errorf("expected default 1s, got %v", result)
+	}
+}
+
+func TestCorrelationConfig_GetMatchingMode(t *testing.T) {
+	tests := []struct {
+		name     string
+		config   CorrelationConfig
+		expected string
+	}{
+		{
+			name:     "one_to_many",
+			config:   CorrelationConfig{Matching: MatchingConfig{Mode: "one_to_many"}},
+			expected: "one_to_many",
+		},
+		{
+			name:     "one_to_one",
+			config:   CorrelationConfig{Matching: MatchingConfig{Mode: "one_to_one"}},
+			expected: "one_to_one",
+		},
+		{
+			name:     "default empty",
+			config:   CorrelationConfig{Matching: MatchingConfig{Mode: ""}},
+			expected: "one_to_many",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := tt.config.GetMatchingMode()
+			if result != tt.expected {
+				t.Errorf("expected %s, got %s", tt.expected, result)
+			}
+		})
+	}
+}
+
+func TestCorrelationConfig_GetApacheAlwaysEmit(t *testing.T) {
+	cfg := CorrelationConfig{
+		OrphanPolicy: OrphanPolicyConfig{ApacheAlwaysEmit: true},
+	}
+
+	if !cfg.GetApacheAlwaysEmit() {
+		t.Error("expected true")
+	}
+}
+
+func TestCorrelationConfig_GetApacheAlwaysEmit_Deprecated(t *testing.T) {
+	cfg := CorrelationConfig{
+		EmitOrphans: true, // deprecated field
+	}
+
+	if !cfg.GetApacheAlwaysEmit() {
+		t.Error("expected true from deprecated field")
+	}
+}
+
+func TestCorrelationConfig_GetNetworkTTLS(t *testing.T) {
+	cfg := CorrelationConfig{
+		TTL: TTLConfig{NetworkTTLS: 60},
+	}
+
+	result := cfg.GetNetworkTTLS()
+	if result != 60 {
+		t.Errorf("expected 60, got %d", result)
+	}
+}
+
+func TestCorrelationConfig_GetNetworkTTLS_Default(t *testing.T) {
+	cfg := CorrelationConfig{
+		TTL: TTLConfig{NetworkTTLS: 0},
+	}
+
+	result := cfg.GetNetworkTTLS()
+	if result != 30 {
+		t.Errorf("expected default 30, got %d", result)
+	}
+}
+
+func TestCorrelationConfig_GetMaxHTTPBufferSize(t *testing.T) {
+	cfg := CorrelationConfig{
+		Buffers: BuffersConfig{MaxHTTPItems: 5000},
+	}
+
+	result := cfg.GetMaxHTTPBufferSize()
+	if result != 5000 {
+		t.Errorf("expected 5000, got %d", result)
+	}
+}
+
+func TestCorrelationConfig_GetMaxHTTPBufferSize_Default(t *testing.T) {
+	cfg := CorrelationConfig{
+		Buffers: BuffersConfig{MaxHTTPItems: 0},
+	}
+
+	result := cfg.GetMaxHTTPBufferSize()
+	if result != 10000 {
+		t.Errorf("expected default 10000, got %d", result)
+	}
+}
+
+func TestCorrelationConfig_GetMaxNetworkBufferSize(t *testing.T) {
+	cfg := CorrelationConfig{
+		Buffers: BuffersConfig{MaxNetworkItems: 15000},
+	}
+
+	result := cfg.GetMaxNetworkBufferSize()
+	if result != 15000 {
+		t.Errorf("expected 15000, got %d", result)
+	}
+}
+
+func TestCorrelationConfig_GetMaxNetworkBufferSize_Default(t *testing.T) {
+	cfg := CorrelationConfig{
+		Buffers: BuffersConfig{MaxNetworkItems: 0},
+	}
+
+	result := cfg.GetMaxNetworkBufferSize()
+	if result != 20000 {
+		t.Errorf("expected default 20000, got %d", result)
+	}
+}
+
+func TestUnixSocketConfig_GetSocketPermissions_Invalid(t *testing.T) {
+	cfg := UnixSocketConfig{
+		SocketPermissions: "invalid",
+	}
+
+	result := cfg.GetSocketPermissions()
+	if result != 0666 {
+		t.Errorf("expected default 0666 for invalid value, got %o", result)
+	}
+}
+
+func TestLoadConfig_Validation(t *testing.T) {
+	// Test config with no inputs
+	content := `
+outputs:
+  file:
+    path: /var/log/test.log
+
+correlation:
+  time_window_s: 1
+`
+
+	tmpDir := t.TempDir()
+	configPath := filepath.Join(tmpDir, "config.yml")
+	if err := os.WriteFile(configPath, []byte(content), 0644); err != nil {
+		t.Fatalf("failed to write config: %v", err)
+	}
+
+	_, err := Load(configPath)
+	if err == nil {
+		t.Error("expected error for config with no inputs")
+	}
+}
+
+func TestLoadConfig_Validation_NoOutputs(t *testing.T) {
+	// Test config with no outputs
+	content := `
+inputs:
+  unix_sockets:
+    - name: http
+      path: /var/run/logcorrelator/http.socket
+
+correlation:
+  time_window_s: 1
+`
+
+	tmpDir := t.TempDir()
+	configPath := filepath.Join(tmpDir, "config.yml")
+	if err := os.WriteFile(configPath, []byte(content), 0644); err != nil {
+		t.Fatalf("failed to write config: %v", err)
+	}
+
+	_, err := Load(configPath)
+	if err == nil {
+		t.Error("expected error for config with no outputs")
+	}
+}
+
+func TestLoadConfig_Validation_DuplicateSocketNames(t *testing.T) {
+	content := `
+inputs:
+  unix_sockets:
+    - name: http
+      path: /var/run/logcorrelator/http.socket
+    - name: http
+      path: /var/run/logcorrelator/other.socket
+
+outputs:
+  file:
+    path: /var/log/test.log
+
+correlation:
+  time_window_s: 1
+`
+
+	tmpDir := t.TempDir()
+	configPath := filepath.Join(tmpDir, "config.yml")
+	if err := os.WriteFile(configPath, []byte(content), 0644); err != nil {
+		t.Fatalf("failed to write config: %v", err)
+	}
+
+	_, err := Load(configPath)
+	if err == nil {
+		t.Error("expected error for duplicate socket names")
+	}
+}
+
+func TestLoadConfig_Validation_DuplicateSocketPaths(t *testing.T) {
+	content := `
+inputs:
+  unix_sockets:
+    - name: http
+      path: /var/run/logcorrelator/http.socket
+    - name: network
+      path: /var/run/logcorrelator/http.socket
+
+outputs:
+  file:
+    path: /var/log/test.log
+
+correlation:
+  time_window_s: 1
+`
+
+	tmpDir := t.TempDir()
+	configPath := filepath.Join(tmpDir, "config.yml")
+	if err := os.WriteFile(configPath, []byte(content), 0644); err != nil {
+		t.Fatalf("failed to write config: %v", err)
+	}
+
+	_, err := Load(configPath)
+	if err == nil {
+		t.Error("expected error for duplicate socket paths")
+	}
+}
+
+func TestLoadConfig_Validation_ClickHouseEnabled(t *testing.T) {
+	content := `
+inputs:
+  unix_sockets:
+    - name: http
+      path: /var/run/logcorrelator/http.socket
+
+outputs:
+  clickhouse:
+    enabled: true
+
+correlation:
+  time_window_s: 1
+`
+
+	tmpDir := t.TempDir()
+	configPath := filepath.Join(tmpDir, "config.yml")
+	if err := os.WriteFile(configPath, []byte(content), 0644); err != nil {
+		t.Fatalf("failed to write config: %v", err)
+	}
+
+	_, err := Load(configPath)
+	if err == nil {
+		t.Error("expected error for ClickHouse enabled without DSN")
+	}
+}
+
+func TestLoadConfig_Validation_ClickHouseEnabled_NoTable(t *testing.T) {
+	content := `
+inputs:
+  unix_sockets:
+    - name: http
+      path: /var/run/logcorrelator/http.socket
+
+outputs:
+  clickhouse:
+    enabled: true
+    dsn: clickhouse://localhost:9000/db
+
+correlation:
+  time_window_s: 1
+`
+
+	tmpDir := t.TempDir()
+	configPath := filepath.Join(tmpDir, "config.yml")
+	if err := os.WriteFile(configPath, []byte(content), 0644); err != nil {
+		t.Fatalf("failed to write config: %v", err)
+	}
+
+	_, err := Load(configPath)
+	if err == nil {
+		t.Error("expected error for ClickHouse enabled without table")
+	}
+}

internal/observability/logger_test.go

@@ -140,3 +140,167 @@ func TestLogger_Name(t *testing.T) {
 		t.Errorf("expected prefix 'myservice', got %s", logger.prefix)
 	}
 }
+
+func TestLogLevel_String(t *testing.T) {
+	tests := []struct {
+		level    LogLevel
+		expected string
+	}{
+		{DEBUG, "DEBUG"},
+		{INFO, "INFO"},
+		{WARN, "WARN"},
+		{ERROR, "ERROR"},
+		{99, "INFO"}, // Unknown level defaults to INFO
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.expected, func(t *testing.T) {
+			result := tt.level.String()
+			if result != tt.expected {
+				t.Errorf("LogLevel(%d).String() = %q, want %q", tt.level, result, tt.expected)
+			}
+		})
+	}
+}
+
+func TestLogger_Warn(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "WARN")
+
+	// WARN should be logged
+	if !logger.ShouldLog(WARN) {
+		t.Error("WARN should be enabled")
+	}
+	logger.Warn("warning message") // Should not panic
+}
+
+func TestLogger_Warnf(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "WARN")
+	logger.Warnf("formatted %s %d", "message", 42) // Should not panic
+}
+
+func TestLogger_Infof(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "INFO")
+	logger.Infof("formatted %s %d", "message", 42) // Should not panic
+}
+
+func TestLogger_Debugf(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "DEBUG")
+	logger.Debugf("formatted %s %d", "message", 42) // Should not panic
+
+	// Debug disabled
+	logger.SetLevel("INFO")
+	logger.Debugf("should not be logged") // Should not panic
+}
+
+func TestLogger_Error_WithError(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "ERROR")
+	testErr := &testError{"test error"}
+	logger.Error("error occurred", testErr) // Should not panic
+}
+
+func TestLogger_Error_WithNilError(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "ERROR")
+	logger.Error("error occurred", nil) // Should not panic
+}
+
+func TestLogger_WithFields_MergesFields(t *testing.T) {
+	logger := NewLogger("test")
+	logger.fields["existing"] = "value"
+
+	fieldsLogger := logger.WithFields(map[string]any{
+		"new": "field",
+	})
+
+	if len(fieldsLogger.fields) != 2 {
+		t.Errorf("expected 2 fields, got %d", len(fieldsLogger.fields))
+	}
+	if _, ok := fieldsLogger.fields["existing"]; !ok {
+		t.Error("expected existing field to be preserved")
+	}
+	if _, ok := fieldsLogger.fields["new"]; !ok {
+		t.Error("expected new field to be added")
+	}
+}
+
+func TestLogger_WithFields_EmptyFields(t *testing.T) {
+	logger := NewLogger("test")
+	fieldsLogger := logger.WithFields(map[string]any{})
+
+	if len(fieldsLogger.fields) != 0 {
+		t.Errorf("expected 0 fields, got %d", len(fieldsLogger.fields))
+	}
+}
+
+func TestLogger_ShouldLog_Concurrent(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "DEBUG")
+
+	done := make(chan bool)
+	for i := 0; i < 10; i++ {
+		go func() {
+			_ = logger.ShouldLog(DEBUG)
+			done <- true
+		}()
+	}
+
+	for i := 0; i < 10; i++ {
+		<-done
+	}
+}
+
+func TestLogger_Log_Concurrent(t *testing.T) {
+	logger := NewLoggerWithLevel("test", "DEBUG")
+
+	done := make(chan bool)
+	for i := 0; i < 10; i++ {
+		go func(n int) {
+			logger.Debugf("message %d", n)
+			done <- true
+		}(i)
+	}
+
+	for i := 0; i < 10; i++ {
+		<-done
+	}
+}
+
+func TestLogger_WithFields_Concurrent(t *testing.T) {
+	logger := NewLogger("test")
+
+	done := make(chan bool)
+	for i := 0; i < 10; i++ {
+		go func(n int) {
+			_ = logger.WithFields(map[string]any{"key": n})
+			done <- true
+		}(i)
+	}
+
+	for i := 0; i < 10; i++ {
+		<-done
+	}
+}
+
+func TestLogger_SetLevel_Concurrent(t *testing.T) {
+	logger := NewLogger("test")
+
+	done := make(chan bool)
+	for i := 0; i < 10; i++ {
+		go func() {
+			logger.SetLevel("DEBUG")
+			logger.SetLevel("INFO")
+			done <- true
+		}()
+	}
+
+	for i := 0; i < 10; i++ {
+		<-done
+	}
+}
+
+// testError implements error for testing
+type testError struct {
+	msg string
+}
+
+func (e *testError) Error() string {
+	return e.msg
+}