cerc-io/ipld-eth-server
12
0
Fork 1
Code Issues 10 Pull Requests 2 Actions Packages Projects Releases 37 Wiki Activity

btc converter, publisher, and indexer unit tests

Browse Source
This commit is contained in:
Ian Norden
2020-02-20 16:14:17 -06:00
parent 8643a7f3b6
commit 642e08a04b
29 changed files with 1286 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/super_node/eth/indexer_test.go
+4 -5
View File
@@ -19,24 +19,23 @@ package eth_test
import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth"
eth2 "github.com/vulcanize/vulcanizedb/pkg/super_node/eth"
"github.com/vulcanize/vulcanizedb/pkg/super_node/eth/mocks"
"github.com/vulcanize/vulcanizedb/pkg/super_node/shared"
)
var _ = Describe("Indexer", func() {
var (
db *postgres.DB
err error
repo *eth2.CIDIndexer
repo *eth.CIDIndexer
)
BeforeEach(func() {
db, err = eth.SetupDB()
db, err = shared.SetupDB()
Expect(err).ToNot(HaveOccurred())
repo = eth2.NewCIDIndexer(db)
repo = eth.NewCIDIndexer(db)
})
AfterEach(func() {
eth.TearDownDB(db)
pkg/super_node/eth/resolver.go
+12 -36
View File
@@ -41,48 +41,24 @@ func (eir *IPLDResolver) Resolve(iplds shared.FetchedIPLDs) (shared.ServerRespon
}
return StreamResponse{
BlockNumber: ipfsBlocks.BlockNumber,
HeadersRlp: eir.ResolveHeaders(ipfsBlocks.Headers),
UnclesRlp: eir.ResolveUncles(ipfsBlocks.Uncles),
TransactionsRlp: eir.ResolveTransactions(ipfsBlocks.Transactions),
ReceiptsRlp: eir.ResolveReceipts(ipfsBlocks.Receipts),
StateNodesRlp: eir.ResolveState(ipfsBlocks.StateNodes),
StorageNodesRlp: eir.ResolveStorage(ipfsBlocks.StorageNodes),
HeadersRlp: eir.resolve(ipfsBlocks.Headers),
UnclesRlp: eir.resolve(ipfsBlocks.Uncles),
TransactionsRlp: eir.resolve(ipfsBlocks.Transactions),
ReceiptsRlp: eir.resolve(ipfsBlocks.Receipts),
StateNodesRlp: eir.resolveState(ipfsBlocks.StateNodes),
StorageNodesRlp: eir.resolveStorage(ipfsBlocks.StorageNodes),
}, nil
}
func (eir *IPLDResolver) ResolveHeaders(iplds []blocks.Block) [][]byte {
headerRlps := make([][]byte, 0, len(iplds))
func (eir *IPLDResolver) resolve(iplds []blocks.Block) [][]byte {
rlps := make([][]byte, 0, len(iplds))
for _, ipld := range iplds {
headerRlps = append(headerRlps, ipld.RawData())
rlps = append(rlps, ipld.RawData())
}
return headerRlps
return rlps
}
func (eir *IPLDResolver) ResolveUncles(iplds []blocks.Block) [][]byte {
uncleRlps := make([][]byte, 0, len(iplds))
for _, ipld := range iplds {
uncleRlps = append(uncleRlps, ipld.RawData())
}
return uncleRlps
}
func (eir *IPLDResolver) ResolveTransactions(iplds []blocks.Block) [][]byte {
trxs := make([][]byte, 0, len(iplds))
for _, ipld := range iplds {
trxs = append(trxs, ipld.RawData())
}
return trxs
}
func (eir *IPLDResolver) ResolveReceipts(iplds []blocks.Block) [][]byte {
rcts := make([][]byte, 0, len(iplds))
for _, ipld := range iplds {
rcts = append(rcts, ipld.RawData())
}
return rcts
}
func (eir *IPLDResolver) ResolveState(iplds map[common.Hash]blocks.Block) map[common.Hash][]byte {
func (eir *IPLDResolver) resolveState(iplds map[common.Hash]blocks.Block) map[common.Hash][]byte {
stateNodes := make(map[common.Hash][]byte, len(iplds))
for key, ipld := range iplds {
stateNodes[key] = ipld.RawData()
@@ -90,7 +66,7 @@ func (eir *IPLDResolver) ResolveState(iplds map[common.Hash]blocks.Block) map[co
return stateNodes
}
func (eir *IPLDResolver) ResolveStorage(iplds map[common.Hash]map[common.Hash]blocks.Block) map[common.Hash]map[common.Hash][]byte {
func (eir *IPLDResolver) resolveStorage(iplds map[common.Hash]map[common.Hash]blocks.Block) map[common.Hash]map[common.Hash][]byte {
storageNodes := make(map[common.Hash]map[common.Hash][]byte)
for stateKey, storageIPLDs := range iplds {
storageNodes[stateKey] = make(map[common.Hash][]byte)
pkg/super_node/eth/retriever.go
+16 -13
View File
@@ -176,18 +176,21 @@ func (ecr *CIDRetriever) RetrieveTxCIDs(tx *sqlx.Tx, txFilter TxFilter, blockNum
log.Debug("retrieving transaction cids for block ", blockNumber)
args := make([]interface{}, 0, 3)
results := make([]TxModel, 0)
pgStr := `SELECT transaction_cids.id, transaction_cids.header_id,
id := 1
pgStr := fmt.Sprintf(`SELECT transaction_cids.id, transaction_cids.header_id,
transaction_cids.tx_hash, transaction_cids.cid,
transaction_cids.dst, transaction_cids.src, transaction_cids.index
FROM eth.transaction_cids INNER JOIN eth.header_cids ON (transaction_cids.header_id = header_cids.id)
WHERE header_cids.block_number = $1`
WHERE header_cids.block_number = $%d`, id)
args = append(args, blockNumber)
id++
if len(txFilter.Dst) > 0 {
pgStr += ` AND transaction_cids.dst = ANY($2::VARCHAR(66)[])`
pgStr += fmt.Sprintf(` AND transaction_cids.dst = ANY($%d::VARCHAR(66)[])`, id)
args = append(args, pq.Array(txFilter.Dst))
id++
}
if len(txFilter.Src) > 0 {
pgStr += ` AND transaction_cids.src = ANY($3::VARCHAR(66)[])`
pgStr += fmt.Sprintf(` AND transaction_cids.src = ANY($%d::VARCHAR(66)[])`, id)
args = append(args, pq.Array(txFilter.Src))
}
return results, tx.Select(&results, pgStr, args...)
@@ -309,26 +312,26 @@ func (ecr *CIDRetriever) RetrieveStateCIDs(tx *sqlx.Tx, stateFilter StateFilter,
func (ecr *CIDRetriever) RetrieveStorageCIDs(tx *sqlx.Tx, storageFilter StorageFilter, blockNumber int64) ([]StorageNodeWithStateKeyModel, error) {
log.Debug("retrieving storage cids for block ", blockNumber)
args := make([]interface{}, 0, 3)
pgStr := `SELECT storage_cids.id, storage_cids.state_id, storage_cids.storage_key,
id := 1
pgStr := fmt.Sprintf(`SELECT storage_cids.id, storage_cids.state_id, storage_cids.storage_key,
storage_cids.leaf, storage_cids.cid, state_cids.state_key FROM eth.storage_cids, eth.state_cids, eth.header_cids
WHERE storage_cids.state_id = state_cids.id
AND state_cids.header_id = header_cids.id
AND header_cids.block_number = $1`
AND header_cids.block_number = $%d`, id)
args = append(args, blockNumber)
id++
addrLen := len(storageFilter.Addresses)
if addrLen > 0 {
keys := make([]string, addrLen)
for i, addr := range storageFilter.Addresses {
keys[i] = crypto.Keccak256Hash(common.HexToAddress(addr).Bytes()).String()
}
pgStr += ` AND state_cids.state_key = ANY($2::VARCHAR(66)[])`
pgStr += fmt.Sprintf(` AND state_cids.state_key = ANY($%d::VARCHAR(66)[])`, id)
args = append(args, pq.Array(keys))
if len(storageFilter.StorageKeys) > 0 {
pgStr += ` AND storage_cids.storage_key = ANY($3::VARCHAR(66)[])`
args = append(args, pq.Array(storageFilter.StorageKeys))
}
} else if len(storageFilter.StorageKeys) > 0 {
pgStr += ` AND storage_cids.storage_key = ANY($2::VARCHAR(66)[])`
id++
}
if len(storageFilter.StorageKeys) > 0 {
pgStr += fmt.Sprintf(` AND storage_cids.storage_key = ANY($%d::VARCHAR(66)[])`, id)
args = append(args, pq.Array(storageFilter.StorageKeys))
}
if !storageFilter.IntermediateNodes {
pkg/super_node/eth/retriever_test.go
+1 -1
View File
@@ -212,7 +212,7 @@ var _ = Describe("Retriever", func() {
)
BeforeEach(func() {
var err error
db, err = eth.SetupDB()
db, err = shared.SetupDB()
Expect(err).ToNot(HaveOccurred())
repo = eth2.NewCIDIndexer(db)
retriever = eth2.NewCIDRetriever(db)
pkg/super_node/eth/test_helpers.go
-11
View File
@@ -19,20 +19,9 @@ package eth
import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/eth/core"
"github.com/vulcanize/vulcanizedb/pkg/eth/datastore/postgres"
)
// SetupDB is use to setup a db for super node tests
func SetupDB() (*postgres.DB, error) {
return postgres.NewDB(config.Database{
Hostname: "localhost",
Name: "vulcanize_testing",
Port: 5432,
}, core.Node{})
}
// TearDownDB is used to tear down the super node dbs after tests
func TearDownDB(db *postgres.DB) {
tx, err := db.Beginx()