Function TestQueueTransactionFlowAsyncPolled

transaction_test.go:4603–4726

(t *testing.T)

}

func TestQueueTransactionFlowAsyncPolled(t *testing.T) {
	// Skip in short mode
	if testing.Short() {
		t.Skip("Skipping queue flow test in short mode")
	}

	ctx := context.Background()
	cnf := &config.Configuration{
		Redis: config.RedisConfig{
			Dns: "localhost:6379", // Ensure Redis is running for queue tests
		},
		DataSource: config.DataSourceConfig{
			Dns: "postgres://postgres:password@localhost:5432/ledgerforge?sslmode=disable", // Ensure Postgres is running
		},
		Queue: config.QueueConfig{ // Ensure queue names are unique if tests run in parallel
			WebhookQueue:        "webhook_queue_test_async_polled",
			IndexQueue:          "index_queue_test_async_polled",
			TransactionQueue:    "transaction_queue_test_async_polled", // Base name for transaction queue
			NumberOfQueues:      1,                                     // Test with one queue for simplicity
			InflightExpiryQueue: "inflight_expiry_queue_test_async_polled",
		},
		Server: config.ServerConfig{
			SecretKey: "test-secret-async-polled",
		},
		Transaction: config.TransactionConfig{
			BatchSize:        100,
			MaxQueueSize:     1000,
			LockDuration:     time.Second * 30,
			IndexQueuePrefix: "test_index_async_polled",
		},
	}
	config.ConfigStore.Store(cnf)

	ds, err := database.NewDataSource(cnf)
	require.NoError(t, err, "Failed to create datasource")

	ledgerforgeInstance, err := NewLedgerForge(ds)
	require.NoError(t, err, "Failed to create LedgerForge instance")

	// Construct the specific transaction queue name
	transactionQueueName := fmt.Sprintf("%s_%d", cnf.Queue.TransactionQueue, 1)

	// Start the test Asynq worker
	cleanupWorker := startTestAsynqWorker(t, cnf, ledgerforgeInstance, transactionQueueName)
	defer cleanupWorker()

	txnRef := "txn_" + model.GenerateUUIDWithSuffix("test_async_polled")

	sourceBalance := &model.Balance{
		Currency: "USD",
		LedgerID: "general_ledger_id",
	}
	destBalance := &model.Balance{
		Currency: "USD",
		LedgerID: "general_ledger_id",
	}

	source, err := ds.CreateBalance(*sourceBalance)