Function TestLineageOutboxProcessor

lineage_test.go:1325–1416

(t *testing.T)

}

func TestLineageOutboxProcessor(t *testing.T) {
	t.Run("Creates processor with default values", func(t *testing.T) {
		mockDS := new(mocks.MockDataSource)
		ledgerforgeInstance := &LedgerForge{datasource: mockDS}

		processor := NewLineageOutboxProcessor(ledgerforgeInstance)

		assert.NotNil(t, processor)
		assert.Equal(t, 100, processor.batchSize)
		assert.Equal(t, 1*time.Second, processor.pollInterval)
		assert.Equal(t, 30*time.Second, processor.lockDuration)
		assert.False(t, processor.IsRunning())
	})

	t.Run("Configures processor with custom values", func(t *testing.T) {
		mockDS := new(mocks.MockDataSource)
		ledgerforgeInstance := &LedgerForge{datasource: mockDS}

		processor := NewLineageOutboxProcessor(ledgerforgeInstance).
			WithBatchSize(50).
			WithPollInterval(5 * time.Second).
			WithLockDuration(60 * time.Second)

		assert.Equal(t, 50, processor.batchSize)
		assert.Equal(t, 5*time.Second, processor.pollInterval)
		assert.Equal(t, 60*time.Second, processor.lockDuration)
	})

	t.Run("Start and stop processor", func(t *testing.T) {
		mockDS := new(mocks.MockDataSource)
		ledgerforgeInstance := &LedgerForge{datasource: mockDS}

		// Mock ClaimPendingOutboxEntries to return empty slice
		mockDS.On("ClaimPendingOutboxEntries", mock.Anything, mock.Anything, mock.Anything).
			Return([]model.LineageOutbox{}, nil).Maybe()

		processor := NewLineageOutboxProcessor(ledgerforgeInstance).
			WithPollInterval(100 * time.Millisecond)

		ctx := context.Background()

		// Start processor
		processor.Start(ctx)
		assert.True(t, processor.IsRunning())

		// Wait a bit for at least one poll cycle
		time.Sleep(150 * time.Millisecond)

		// Stop processor
		processor.Stop()
		assert.False(t, processor.IsRunning())
	})

	t.Run("Processes batch of entries", func(t *testing.T) {
		mockDS := new(mocks.MockDataSource)

		txn := &model.Transaction{
			TransactionID: "txn_batch_1",