crawlab/core/task/handler/service_robustness_test.go

package handler

import (
	"context"
	"sync"
	"testing"
	"time"

	"github.com/crawlab-team/crawlab/core/utils"
	"go.mongodb.org/mongo-driver/bson/primitive"
)

// TestService_GracefulShutdown tests proper service shutdown
func TestService_GracefulShutdown(t *testing.T) {
	svc := &Service{
		fetchInterval:  100 * time.Millisecond,
		reportInterval: 100 * time.Millisecond,
		mu:             sync.RWMutex{},
		runners:        sync.Map{},
		Logger:         utils.NewLogger("TestService"),
	}

	// Initialize context
	svc.ctx, svc.cancel = context.WithCancel(context.Background())

	// Initialize tickers
	svc.fetchTicker = time.NewTicker(svc.fetchInterval)
	svc.reportTicker = time.NewTicker(svc.reportInterval)

	// Start background goroutines
	svc.wg.Add(2)
	go svc.testFetchAndRunTasks() // Mock version
	go svc.testReportStatus()     // Mock version

	// Let it run for a short time
	time.Sleep(200 * time.Millisecond)

	// Test graceful shutdown
	svc.Stop()

	t.Log("✅ Service shutdown completed gracefully")
}

// Mock versions for testing without dependencies
func (svc *Service) testFetchAndRunTasks() {
	defer svc.wg.Done()
	defer func() {
		if r := recover(); r != nil {
			svc.Errorf("testFetchAndRunTasks panic recovered: %v", r)
		}
	}()

	for {
		select {
		case <-svc.ctx.Done():
			svc.Infof("testFetchAndRunTasks stopped by context")
			return
		case <-svc.fetchTicker.C:
			// Mock fetch operation
			svc.Debugf("Mock fetch operation")
		}
	}
}

func (svc *Service) testReportStatus() {
	defer svc.wg.Done()
	defer func() {
		if r := recover(); r != nil {
			svc.Errorf("testReportStatus panic recovered: %v", r)
		}
	}()

	for {
		select {
		case <-svc.ctx.Done():
			svc.Infof("testReportStatus stopped by context")
			return
		case <-svc.reportTicker.C:
			// Mock status update
			svc.Debugf("Mock status update")
		}
	}
}

// TestService_ConcurrentAccess tests thread safety
func TestService_ConcurrentAccess(t *testing.T) {
	svc := &Service{
		mu:      sync.RWMutex{},
		runners: sync.Map{},
		Logger:  utils.NewLogger("TestService"),
	}

	// Initialize context
	svc.ctx, svc.cancel = context.WithCancel(context.Background())
	defer svc.cancel()

	// Test concurrent runner management
	var wg sync.WaitGroup
	numGoroutines := 50

	// Mock runner for testing
	mockRunner := &mockTaskRunner{id: primitive.NewObjectID()}

	// Concurrent adds
	for i := 0; i < numGoroutines; i++ {
		wg.Add(1)
		go func(index int) {
			defer wg.Done()
			taskId := primitive.NewObjectID()
			svc.addRunner(taskId, mockRunner)

			// Brief pause
			time.Sleep(time.Millisecond)

			// Test get runner
			_, err := svc.getRunner(taskId)
			if err != nil {
				t.Errorf("Failed to get runner: %v", err)
			}

			// Delete runner
			svc.deleteRunner(taskId)
		}(i)
	}

	wg.Wait()
	t.Log("✅ Concurrent access test completed successfully")
}

// TestService_ErrorHandling tests error recovery
func TestService_ErrorHandling(t *testing.T) {
	svc := &Service{
		mu:      sync.RWMutex{},
		runners: sync.Map{},
		Logger:  utils.NewLogger("TestService"),
	}

	// Test getting non-existent runner
	_, err := svc.getRunner(primitive.NewObjectID())
	if err == nil {
		t.Error("Expected error for non-existent runner")
	}

	// Test adding invalid runner type
	taskId := primitive.NewObjectID()
	svc.runners.Store(taskId, "invalid-type")

	_, err = svc.getRunner(taskId)
	if err == nil {
		t.Error("Expected error for invalid runner type")
	}

	t.Log("✅ Error handling test completed successfully")
}

// TestService_ResourceCleanup tests proper resource cleanup
func TestService_ResourceCleanup(t *testing.T) {
	svc := &Service{
		mu:      sync.RWMutex{},
		runners: sync.Map{},
		Logger:  utils.NewLogger("TestService"),
	}

	// Initialize context and tickers
	svc.ctx, svc.cancel = context.WithCancel(context.Background())
	svc.fetchTicker = time.NewTicker(100 * time.Millisecond)
	svc.reportTicker = time.NewTicker(100 * time.Millisecond)

	// Add some mock runners
	for i := 0; i < 5; i++ {
		taskId := primitive.NewObjectID()
		mockRunner := &mockTaskRunner{id: taskId}
		svc.addRunner(taskId, mockRunner)
	}

	// Verify runners exist
	runnerCount := 0
	svc.runners.Range(func(key, value interface{}) bool {
		runnerCount++
		return true
	})
	if runnerCount != 5 {
		t.Errorf("Expected 5 runners, got %d", runnerCount)
	}

	// Test cleanup
	svc.stopAllRunners()

	// Verify cleanup (runners should still exist but be marked for cancellation)
	// In a real scenario, runners would remove themselves after cancellation
	t.Log("✅ Resource cleanup test completed successfully")
}

// Mock task runner for testing
type mockTaskRunner struct {
	id primitive.ObjectID
}

func (r *mockTaskRunner) Init() error {
	return nil
}

func (r *mockTaskRunner) GetTaskId() primitive.ObjectID {
	return r.id
}

func (r *mockTaskRunner) Run() error {
	return nil
}

func (r *mockTaskRunner) Cancel(force bool) error {
	return nil
}

func (r *mockTaskRunner) SetSubscribeTimeout(timeout time.Duration) {
	// Mock implementation
}