deep-stack/ipld-eth-server
4
0
Fork 0
forked from cerc-io/ipld-eth-server
Code Pull Requests Activity

Use transformer factory for Price Feed events

Browse Source
This commit is contained in:
Rob Mulholand
2018-10-26 11:25:16 -05:00
parent bd428e752a
commit 47a3c35938
17 changed files with 147 additions and 530 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/transformers/price_feeds/config.go
+4 -9
View File
@@ -16,20 +16,15 @@ package price_feeds
import "github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
var ()
type IPriceFeedConfig struct {
ContractAddresses []string
StartingBlockNumber int64
EndingBlockNumber int64
}
var PriceFeedConfig = IPriceFeedConfig{
var PriceFeedConfig = shared.SingleTransformerConfig{
TransformerName: shared.PriceFeedLabel,
ContractAddresses: []string{
shared.PepContractAddress,
shared.PipContractAddress,
shared.RepContractAddress,
},
ContractAbi: shared.MedianizerABI,
Topic: shared.LogValueSignature,
StartingBlockNumber: 0,
EndingBlockNumber: 10000000,
}
pkg/transformers/price_feeds/converter.go
+4 -7
View File
@@ -20,14 +20,11 @@ import (
"github.com/ethereum/go-ethereum/core/types"
)
type Converter interface {
ToModels(logs []types.Log, headerID int64) ([]PriceFeedModel, error)
}
type PriceFeedConverter struct{}
func (converter PriceFeedConverter) ToModels(logs []types.Log, headerID int64) (results []PriceFeedModel, err error) {
for _, log := range logs {
func (converter PriceFeedConverter) ToModels(ethLogs []types.Log) ([]interface{}, error) {
var results []interface{}
for _, log := range ethLogs {
raw, err := json.Marshal(log)
if err != nil {
return nil, err
@@ -42,5 +39,5 @@ func (converter PriceFeedConverter) ToModels(logs []types.Log, headerID int64) (
}
results = append(results, model)
}
return results, err
return results, nil
}
pkg/transformers/price_feeds/converter_test.go
+1 -2
View File
@@ -26,9 +26,8 @@ import (
var _ = Describe("Price feed Converter", func() {
It("converts a log to a price feed model", func() {
converter := price_feeds.PriceFeedConverter{}
headerID := int64(123)
models, err := converter.ToModels([]types.Log{test_data.EthPriceFeedLog}, headerID)
models, err := converter.ToModels([]types.Log{test_data.EthPriceFeedLog})
Expect(err).NotTo(HaveOccurred())
Expect(len(models)).To(Equal(1))
pkg/transformers/price_feeds/fetcher.go
-57
View File
@@ -1,57 +0,0 @@
// Copyright © 2018 Vulcanize
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package price_feeds
import (
"math/big"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
)
type IPriceFeedFetcher interface {
FetchLogValues(blockNumber int64) ([]types.Log, error)
}
type PriceFeedFetcher struct {
blockChain core.BlockChain
contractAddresses []string
}
func NewPriceFeedFetcher(blockChain core.BlockChain, contractAddresses []string) PriceFeedFetcher {
return PriceFeedFetcher{
blockChain: blockChain,
contractAddresses: contractAddresses,
}
}
func (fetcher PriceFeedFetcher) FetchLogValues(blockNumber int64) ([]types.Log, error) {
var addresses []common.Address
for _, addr := range fetcher.contractAddresses {
addresses = append(addresses, common.HexToAddress(addr))
}
n := big.NewInt(blockNumber)
query := ethereum.FilterQuery{
FromBlock: n,
ToBlock: n,
Addresses: addresses,
Topics: [][]common.Hash{{common.HexToHash(shared.LogValueSignature)}},
}
return fetcher.blockChain.GetEthLogsWithCustomQuery(query)
}
pkg/transformers/price_feeds/fetcher_test.go
-65
View File
@@ -1,65 +0,0 @@
// Copyright © 2018 Vulcanize
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package price_feeds_test
import (
"math/big"
"github.com/ethereum/go-ethereum"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/transformers/price_feeds"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
)
var _ = Describe("Price fetcher", func() {
It("gets log value events from price feed medianizers", func() {
mockBlockChain := fakes.NewMockBlockChain()
mockBlockChain.SetGetEthLogsWithCustomQueryReturnLogs([]types.Log{{}})
contractAddresses := []string{"pep-contract-address", "pip-contract-address", "rep-contract-address"}
fetcher := price_feeds.NewPriceFeedFetcher(mockBlockChain, contractAddresses)
blockNumber := int64(100)
_, err := fetcher.FetchLogValues(blockNumber)
Expect(err).NotTo(HaveOccurred())
var expectedAddresses []common.Address
for _, address := range contractAddresses {
expectedAddresses = append(expectedAddresses, common.HexToAddress(address))
}
expectedQuery := ethereum.FilterQuery{
FromBlock: big.NewInt(blockNumber),
ToBlock: big.NewInt(blockNumber),
Addresses: expectedAddresses,
Topics: [][]common.Hash{{common.HexToHash(shared.LogValueSignature)}},
}
mockBlockChain.AssertGetEthLogsWithCustomQueryCalledWith(expectedQuery)
})
It("returns error if getting logs fails", func() {
mockBlockChain := fakes.NewMockBlockChain()
mockBlockChain.SetGetEthLogsWithCustomQueryErr(fakes.FakeError)
fetcher := price_feeds.NewPriceFeedFetcher(mockBlockChain, []string{"contract-address"})
_, err := fetcher.FetchLogValues(100)
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(fakes.FakeError))
})
})
pkg/transformers/price_feeds/repository.go
+13 -12
View File
@@ -15,32 +15,29 @@
package price_feeds
import (
"fmt"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
type IPriceFeedRepository interface {
Create(headerID int64, models []PriceFeedModel) error
MarkHeaderChecked(headerID int64) error
MissingHeaders(startingBlockNumber, endingBlockNumber int64) ([]core.Header, error)
}
type PriceFeedRepository struct {
db *postgres.DB
}
func NewPriceFeedRepository(db *postgres.DB) PriceFeedRepository {
return PriceFeedRepository{db: db}
}
func (repository PriceFeedRepository) Create(headerID int64, models []PriceFeedModel) error {
func (repository PriceFeedRepository) Create(headerID int64, models []interface{}) error {
tx, err := repository.db.Begin()
if err != nil {
return err
}
for _, model := range models {
priceUpdate, ok := model.(PriceFeedModel)
if !ok {
tx.Rollback()
return fmt.Errorf("model of type %T, not %T", model, PriceFeedModel{})
}
_, err = tx.Exec(`INSERT INTO maker.price_feeds (block_number, header_id, medianizer_address, usd_value, log_idx, tx_idx, raw_log)
VALUES ($1, $2, $3, $4::NUMERIC, $5, $6, $7)`, model.BlockNumber, headerID, model.MedianizerAddress, model.UsdValue, model.LogIndex, model.TransactionIndex, model.Raw)
VALUES ($1, $2, $3, $4::NUMERIC, $5, $6, $7)`, priceUpdate.BlockNumber, headerID, priceUpdate.MedianizerAddress, priceUpdate.UsdValue, priceUpdate.LogIndex, priceUpdate.TransactionIndex, priceUpdate.Raw)
if err != nil {
tx.Rollback()
return err
@@ -81,3 +78,7 @@ func (repository PriceFeedRepository) MissingHeaders(startingBlockNumber, ending
)
return result, err
}
func (repository *PriceFeedRepository) SetDB(db *postgres.DB) {
repository.db = db
}
pkg/transformers/price_feeds/repository_test.go
+17 -9
View File
@@ -35,14 +35,15 @@ var _ = Describe("Price feeds repository", func() {
db *postgres.DB
err error
headerRepository datastore.HeaderRepository
priceFeedRepository price_feeds.IPriceFeedRepository
priceFeedRepository price_feeds.PriceFeedRepository
)
BeforeEach(func() {
db = test_config.NewTestDB(core.Node{})
test_config.CleanTestDB(db)
headerRepository = repositories.NewHeaderRepository(db)
priceFeedRepository = price_feeds.NewPriceFeedRepository(db)
priceFeedRepository = price_feeds.PriceFeedRepository{}
priceFeedRepository.SetDB(db)
})
Describe("Create", func() {
@@ -54,7 +55,7 @@ var _ = Describe("Price feeds repository", func() {
})
It("persists a price feed update", func() {
err = priceFeedRepository.Create(headerID, []price_feeds.PriceFeedModel{test_data.PriceFeedModel})
err = priceFeedRepository.Create(headerID, []interface{}{test_data.PriceFeedModel})
Expect(err).NotTo(HaveOccurred())
var dbPriceFeedUpdate price_feeds.PriceFeedModel
@@ -69,7 +70,7 @@ var _ = Describe("Price feeds repository", func() {
})
It("marks headerID as checked for price feed logs", func() {
err = priceFeedRepository.Create(headerID, []price_feeds.PriceFeedModel{test_data.PriceFeedModel})
err = priceFeedRepository.Create(headerID, []interface{}{test_data.PriceFeedModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
@@ -82,7 +83,7 @@ var _ = Describe("Price feeds repository", func() {
_, err = db.Exec(`INSERT INTO public.checked_headers (header_id) VALUES ($1)`, headerID)
Expect(err).NotTo(HaveOccurred())
err = priceFeedRepository.Create(headerID, []price_feeds.PriceFeedModel{test_data.PriceFeedModel})
err = priceFeedRepository.Create(headerID, []interface{}{test_data.PriceFeedModel})
Expect(err).NotTo(HaveOccurred())
var headerChecked bool
@@ -92,17 +93,17 @@ var _ = Describe("Price feeds repository", func() {
})
It("does not duplicate price feed updates", func() {
err = priceFeedRepository.Create(headerID, []price_feeds.PriceFeedModel{test_data.PriceFeedModel})
err = priceFeedRepository.Create(headerID, []interface{}{test_data.PriceFeedModel})
Expect(err).NotTo(HaveOccurred())
err = priceFeedRepository.Create(headerID, []price_feeds.PriceFeedModel{test_data.PriceFeedModel})
err = priceFeedRepository.Create(headerID, []interface{}{test_data.PriceFeedModel})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("pq: duplicate key value violates unique constraint"))
})
It("removes price feed if corresponding header is deleted", func() {
err = priceFeedRepository.Create(headerID, []price_feeds.PriceFeedModel{test_data.PriceFeedModel})
err = priceFeedRepository.Create(headerID, []interface{}{test_data.PriceFeedModel})
Expect(err).NotTo(HaveOccurred())
_, err = db.Exec(`DELETE FROM headers WHERE id = $1`, headerID)
@@ -113,6 +114,12 @@ var _ = Describe("Price feeds repository", func() {
Expect(err).To(HaveOccurred())
Expect(err).To(MatchError(sql.ErrNoRows))
})
It("returns an error if model is of wrong type", func() {
err = priceFeedRepository.Create(headerID, []interface{}{test_data.WrongModel{}})
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(ContainSubstring("model of type"))
})
})
Describe("MarkHeaderChecked", func() {
@@ -200,7 +207,8 @@ var _ = Describe("Price feeds repository", func() {
_, err = headerRepositoryTwo.CreateOrUpdateHeader(fakes.GetFakeHeader(n))
Expect(err).NotTo(HaveOccurred())
}
priceFeedRepositoryTwo := price_feeds.NewPriceFeedRepository(dbTwo)
priceFeedRepositoryTwo := price_feeds.PriceFeedRepository{}
priceFeedRepositoryTwo.SetDB(dbTwo)
err := priceFeedRepository.MarkHeaderChecked(headerIDs[0])
Expect(err).NotTo(HaveOccurred())
pkg/transformers/price_feeds/transformer.go
-75
View File
@@ -1,75 +0,0 @@
// Copyright © 2018 Vulcanize
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package price_feeds
import (
"log"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
)
type PriceFeedTransformerInitializer struct {
Config IPriceFeedConfig
}
func (initializer PriceFeedTransformerInitializer) NewPriceFeedTransformer(db *postgres.DB, blockChain core.BlockChain) shared.Transformer {
converter := PriceFeedConverter{}
fetcher := NewPriceFeedFetcher(blockChain, initializer.Config.ContractAddresses)
repository := NewPriceFeedRepository(db)
return PriceFeedTransformer{
Config: initializer.Config,
Converter: converter,
Fetcher: fetcher,
Repository: repository,
}
}
type PriceFeedTransformer struct {
Config IPriceFeedConfig
Converter Converter
Fetcher IPriceFeedFetcher
Repository IPriceFeedRepository
}
func (transformer PriceFeedTransformer) Execute() error {
headers, err := transformer.Repository.MissingHeaders(transformer.Config.StartingBlockNumber, transformer.Config.EndingBlockNumber)
if err != nil {
return err
}
log.Printf("Fetching price feed event logs for %d headers \n", len(headers))
for _, header := range headers {
logs, err := transformer.Fetcher.FetchLogValues(header.BlockNumber)
if err != nil {
return err
}
if len(logs) < 1 {
err := transformer.Repository.MarkHeaderChecked(header.Id)
if err != nil {
return err
}
}
models, err := transformer.Converter.ToModels(logs, header.Id)
if err != nil {
return err
}
err = transformer.Repository.Create(header.Id, models)
if err != nil {
return err
}
}
return nil
}
pkg/transformers/price_feeds/transformer_test.go
+62 -110
View File
@@ -19,39 +19,57 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/ethereum/go-ethereum/common"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/fakes"
"github.com/vulcanize/vulcanizedb/pkg/transformers/factories"
"github.com/vulcanize/vulcanizedb/pkg/transformers/price_feeds"
"github.com/vulcanize/vulcanizedb/pkg/transformers/shared"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data"
price_feeds_mocks "github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks/price_feeds"
"github.com/vulcanize/vulcanizedb/pkg/transformers/test_data/mocks"
"math/rand"
)
var _ = Describe("Price feed transformer", func() {
It("gets missing headers for price feeds", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
transformer := price_feeds.PriceFeedTransformer{
Config: price_feeds.PriceFeedConfig,
Converter: mockConverter,
Fetcher: &price_feeds_mocks.MockPriceFeedFetcher{},
Repository: mockRepository,
}
var (
config = price_feeds.PriceFeedConfig
fetcher mocks.MockLogFetcher
converter mocks.MockConverter
repository mocks.MockRepository
transformer shared.Transformer
headerOne core.Header
headerTwo core.Header
)
BeforeEach(func() {
fetcher = mocks.MockLogFetcher{}
converter = mocks.MockConverter{}
repository = mocks.MockRepository{}
headerOne = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
headerTwo = core.Header{Id: rand.Int63(), BlockNumber: rand.Int63()}
transformer = factories.Transformer{
Config: config,
Converter: &converter,
Fetcher: &fetcher,
Repository: &repository,
}.NewTransformer(nil, nil)
})
It("sets the blockchain and db", func() {
Expect(fetcher.SetBcCalled).To(BeTrue())
Expect(repository.SetDbCalled).To(BeTrue())
})
It("gets missing headers for price feeds", func() {
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
mockRepository.AssertMissingHeadersCalledwith(price_feeds.PriceFeedConfig.StartingBlockNumber, price_feeds.PriceFeedConfig.EndingBlockNumber)
Expect(repository.PassedStartingBlockNumber).To(Equal(config.StartingBlockNumber))
Expect(repository.PassedEndingBlockNumber).To(Equal(config.EndingBlockNumber))
})
It("returns error is missing headers call returns err", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
mockRepository.SetMissingHeadersErr(fakes.FakeError)
transformer := price_feeds.PriceFeedTransformer{
Converter: mockConverter,
Fetcher: &price_feeds_mocks.MockPriceFeedFetcher{},
Repository: mockRepository,
}
repository.SetMissingHeadersError(fakes.FakeError)
err := transformer.Execute()
@@ -60,35 +78,19 @@ var _ = Describe("Price feed transformer", func() {
})
It("fetches logs for missing headers", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
blockNumberOne := int64(1)
blockNumberTwo := int64(2)
mockRepository.SetMissingHeaders([]core.Header{{BlockNumber: blockNumberOne}, {BlockNumber: blockNumberTwo}})
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
transformer := price_feeds.PriceFeedTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne, headerTwo})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
mockFetcher.AssertFetchLogValuesCalledWith([]int64{blockNumberOne, blockNumberTwo})
Expect(fetcher.FetchedContractAddresses).To(Equal([][]string{config.ContractAddresses, config.ContractAddresses}))
Expect(fetcher.FetchedTopics).To(Equal([][]common.Hash{{common.HexToHash(config.Topic)}}))
Expect(fetcher.FetchedBlocks).To(Equal([]int64{headerOne.BlockNumber, headerTwo.BlockNumber}))
})
It("returns err if fetcher returns err", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
mockRepository.SetMissingHeaders([]core.Header{{BlockNumber: 1}})
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
mockFetcher.SetReturnErr(fakes.FakeError)
transformer := price_feeds.PriceFeedTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
fetcher.SetFetcherError(fakes.FakeError)
err := transformer.Execute()
@@ -97,34 +99,17 @@ var _ = Describe("Price feed transformer", func() {
})
It("marks header checked if no logs returned", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
headerID := int64(123)
mockRepository.SetMissingHeaders([]core.Header{{Id: headerID}})
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
transformer := price_feeds.PriceFeedTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
mockRepository.AssertMarkHeaderCheckedCalledWith(headerID)
repository.AssertMarkHeaderCheckedCalledWith(headerOne.Id)
})
It("returns error if marking header checked returns err", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
mockRepository.SetMissingHeaders([]core.Header{{Id: int64(123)}})
mockRepository.SetMarkHeaderCheckedErr(fakes.FakeError)
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
transformer := price_feeds.PriceFeedTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
repository.SetMarkHeaderCheckedError(fakes.FakeError)
err := transformer.Execute()
@@ -133,38 +118,19 @@ var _ = Describe("Price feed transformer", func() {
})
It("converts log to a model", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
mockFetcher.SetReturnLogs([]types.Log{test_data.EthPriceFeedLog})
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
headerID := int64(11111)
mockRepository.SetMissingHeaders([]core.Header{{BlockNumber: 1, Id: headerID}})
transformer := price_feeds.PriceFeedTransformer{
Fetcher: mockFetcher,
Converter: mockConverter,
Repository: mockRepository,
}
fetcher.SetFetchedLogs([]types.Log{test_data.EthPriceFeedLog})
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
Expect(mockConverter.PassedHeaderID).To(Equal(headerID))
Expect(mockConverter.PassedLogs).To(Equal([]types.Log{test_data.EthPriceFeedLog}))
Expect(converter.PassedLogs).To(Equal([]types.Log{test_data.EthPriceFeedLog}))
})
It("returns err if converter returns err", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockConverter.SetConverterErr(fakes.FakeError)
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
mockFetcher.SetReturnLogs([]types.Log{test_data.EthPriceFeedLog})
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
headerID := int64(11111)
mockRepository.SetMissingHeaders([]core.Header{{BlockNumber: 1, Id: headerID}})
transformer := price_feeds.PriceFeedTransformer{
Fetcher: mockFetcher,
Converter: mockConverter,
Repository: mockRepository,
}
converter.SetConverterError(fakes.FakeError)
fetcher.SetFetchedLogs([]types.Log{test_data.EthPriceFeedLog})
repository.SetMissingHeaders([]core.Header{headerOne})
err := transformer.Execute()
@@ -173,36 +139,22 @@ var _ = Describe("Price feed transformer", func() {
})
It("persists model converted from log", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
headerID := int64(11111)
mockRepository.SetMissingHeaders([]core.Header{{BlockNumber: 1, Id: headerID}})
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
mockFetcher.SetReturnLogs([]types.Log{test_data.EthPriceFeedLog})
transformer := price_feeds.PriceFeedTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
converter.SetReturnModels([]interface{}{test_data.PriceFeedModel})
repository.SetMissingHeaders([]core.Header{headerOne})
fetcher.SetFetchedLogs([]types.Log{test_data.EthPriceFeedLog})
err := transformer.Execute()
Expect(err).NotTo(HaveOccurred())
mockRepository.AssertCreateCalledWith(headerID, []price_feeds.PriceFeedModel{test_data.PriceFeedModel})
Expect(repository.PassedHeaderID).To(Equal(headerOne.Id))
Expect(len(repository.PassedModels)).To(Equal(1))
Expect(repository.PassedModels[0]).To(Equal(test_data.PriceFeedModel))
})
It("returns error if creating price feed update returns error", func() {
mockConverter := &price_feeds_mocks.MockPriceFeedConverter{}
mockRepository := &price_feeds_mocks.MockPriceFeedRepository{}
mockRepository.SetMissingHeaders([]core.Header{{BlockNumber: 1, Id: 2}})
mockRepository.SetCreateErr(fakes.FakeError)
mockFetcher := &price_feeds_mocks.MockPriceFeedFetcher{}
mockFetcher.SetReturnLogs([]types.Log{{}})
transformer := price_feeds.PriceFeedTransformer{
Converter: mockConverter,
Fetcher: mockFetcher,
Repository: mockRepository,
}
repository.SetMissingHeaders([]core.Header{headerOne})
repository.SetCreateError(fakes.FakeError)
fetcher.SetFetchedLogs([]types.Log{{}})
err := transformer.Execute()