diff --git a/internal/cdptest/ledgerlib.go b/internal/cdptest/ledgerlib.go
new file mode 100644
index 00000000..c6f99642
--- /dev/null
+++ b/internal/cdptest/ledgerlib.go
@@ -0,0 +1,70 @@
+package cdptest
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/stellar/go/xdr"
+	"github.com/stellar/stellar-etl/internal/utils"
+)
+
+func LedgerSequence(lcm xdr.LedgerCloseMeta) (*uint32, error) {
+	ledgerSequence := lcm.LedgerSequence()
+	return &ledgerSequence, nil
+}
+
+func CloseTime(lcm xdr.LedgerCloseMeta) (*time.Time, error) {
+	ledgerHeader := lcm.LedgerHeaderHistoryEntry()
+	closeTime, err := utils.TimePointToUTCTimeStamp(ledgerHeader.Header.ScpValue.CloseTime)
+	if err != nil {
+		return nil, err
+	}
+
+	return &closeTime, nil
+}
+
+func BaseFee(lcm xdr.LedgerCloseMeta) (*uint32, error) {
+	ledgerHeader := lcm.LedgerHeaderHistoryEntry()
+	baseFee := uint32(ledgerHeader.Header.BaseFee)
+	return &baseFee, nil
+}
+
+func BaseReserve(lcm xdr.LedgerCloseMeta) (*uint32, error) {
+	ledgerHeader := lcm.LedgerHeaderHistoryEntry()
+	baseReserve := uint32(ledgerHeader.Header.BaseReserve)
+	return &baseReserve, nil
+}
+
+func SorobanFeeWrite1Kb(lcm xdr.LedgerCloseMeta) (*int64, error) {
+	switch lcm.V {
+	case 0:
+		return nil, nil
+	case 1:
+		lcmV1Ext := lcm.MustV1().Ext
+		switch lcmV1Ext.V {
+		case 0:
+			return nil, nil
+		case 1:
+			ext := lcmV1Ext.MustV1()
+			sorobanFreeWrite1Kb := int64(ext.SorobanFeeWrite1Kb)
+			return &sorobanFreeWrite1Kb, nil
+		default:
+			panic(fmt.Errorf("unsupported LedgerCloseMeta.V1.Ext.V: %d", lcmV1Ext.V))
+		}
+	default:
+		panic(fmt.Errorf("unsupported LedgerCloseMeta.V: %d", lcm.V))
+	}
+}
+
+func TotalByteSizeOfBucketList(lcm xdr.LedgerCloseMeta) (*uint64, error) {
+	switch lcm.V {
+	case 0:
+		return nil, nil
+	case 1:
+		lcmV1 := lcm.MustV1()
+		totalByteSizeOfBucketList := uint64(lcmV1.TotalByteSizeOfBucketList)
+		return &totalByteSizeOfBucketList, nil
+	default:
+		panic(fmt.Sprintf("Unsupported LedgerCloseMeta.V: %d", lcm.V))
+	}
+}
diff --git a/internal/cdptest/ledgersprocessor.json b/internal/cdptest/ledgersprocessor.json
new file mode 100644
index 00000000..0988a051
--- /dev/null
+++ b/internal/cdptest/ledgersprocessor.json
@@ -0,0 +1,3 @@
+{
+    "functions": ["LedgerSequence", "CloseTime", "BaseFee", "SorobanFeeWrite1Kb"]
+}
\ No newline at end of file
diff --git a/internal/cdptest/transactionlib.go b/internal/cdptest/transactionlib.go
new file mode 100644
index 00000000..cacfa58e
--- /dev/null
+++ b/internal/cdptest/transactionlib.go
@@ -0,0 +1,31 @@
+package cdptest
+
+import (
+	"github.com/stellar/go/ingest"
+	"github.com/stellar/go/xdr"
+	"github.com/stellar/stellar-etl/internal/utils"
+)
+
+func TransactionHash(tx ingest.LedgerTransaction) (*string, error) {
+	transactionHash := utils.HashToHexString(tx.Result.TransactionHash)
+	return &transactionHash, nil
+}
+
+func Account(tx ingest.LedgerTransaction) (*string, error) {
+	account, err := utils.GetAccountAddressFromMuxedAccount(tx.Envelope.SourceAccount())
+	if err != nil {
+		return nil, err
+	}
+
+	return &account, nil
+
+}
+
+func TransactionEnvelope(tx ingest.LedgerTransaction) (*string, error) {
+	transactionEnvelope, err := xdr.MarshalBase64(tx.Envelope)
+	if err != nil {
+		return nil, err
+	}
+
+	return &transactionEnvelope, nil
+}
diff --git a/internal/cdptest/transactionprocessor.json b/internal/cdptest/transactionprocessor.json
new file mode 100644
index 00000000..5cf89fce
--- /dev/null
+++ b/internal/cdptest/transactionprocessor.json
@@ -0,0 +1,3 @@
+{
+    "functions": ["LedgerSequence", "CloseTime", "TransactionHash", "Account", "TransactionEnvelope"]
+}
\ No newline at end of file
diff --git a/internal/cdptestprocessor/genericprocessor.go b/internal/cdptestprocessor/genericprocessor.go
new file mode 100644
index 00000000..2f76d8e6
--- /dev/null
+++ b/internal/cdptestprocessor/genericprocessor.go
@@ -0,0 +1,81 @@
+package cdptest
+
+import (
+	"io"
+	"maps"
+
+	"github.com/stellar/go/ingest"
+	"github.com/stellar/go/xdr"
+	"github.com/stellar/stellar-etl/internal/cdptest"
+)
+
+var ledgerFunctionMap = map[string]interface{}{
+	"LedgerSequence":            cdptest.LedgerSequence,
+	"CloseTime":                 cdptest.CloseTime,
+	"BaseFee":                   cdptest.BaseFee,
+	"BaseReserve":               cdptest.BaseReserve,
+	"SorobanFeeWrite1Kb":        cdptest.SorobanFeeWrite1Kb,
+	"TotalByteSizeOfBucketList": cdptest.TotalByteSizeOfBucketList,
+}
+
+var transactionFunctionMap = map[string]interface{}{
+	"TransactionHash":     cdptest.TransactionHash,
+	"Account":             cdptest.Account,
+	"TransactionEnvelope": cdptest.TransactionEnvelope,
+}
+
+func GenericProcessor(lcm xdr.LedgerCloseMeta, processorJSON string) ([]map[string]interface{}, error) {
+	// result can be any number of key:val
+	var ledgerResult map[string]interface{}
+
+	// Get the functions from the *processor.json file
+	funcs := make(map[string][]string)
+
+	// LedgerFunctions are single value
+	for _, funcName := range funcs["functions"] {
+		fn, ok := ledgerFunctionMap[funcName]
+		if ok {
+			f := fn.(func(xdr.LedgerCloseMeta) interface{})
+			ledgerResult[funcName] = f(lcm)
+		}
+	}
+
+	hasTx := false
+	for _, key := range funcs["functions"] {
+		_, ok := transactionFunctionMap[key]
+		if ok {
+			hasTx = true
+		}
+	}
+
+	if !hasTx {
+		var results []map[string]interface{}
+		results = append(results, ledgerResult)
+		return results, nil
+	}
+
+	// There can be multiple tx in LCM
+	var transactionResult []map[string]interface{}
+
+	txReader, _ := ingest.NewLedgerTransactionReaderFromLedgerCloseMeta("password", lcm)
+	for {
+		tx, err := txReader.Read()
+		if err == io.EOF {
+			break
+		}
+
+		var result map[string]interface{}
+		for _, funcName := range funcs["functions"] {
+
+			fn, ok := transactionFunctionMap[funcName]
+			if ok {
+				f := fn.(func(ingest.LedgerTransaction) interface{})
+				result[funcName] = f(tx)
+			}
+		}
+		maps.Copy(result, ledgerResult)
+		transactionResult = append(transactionResult, result)
+	}
+
+	return transactionResult, nil
+}