@@ -77,6 +77,101 @@ class ReadRpcJsonSpec extends Specification {
|
||||
reader.getType("".bytes)
|
||||
then:
|
||||
thrown(RpcException)
|
||||
}
|
||||
|
||||
def "Parse basic"() {
|
||||
when:
|
||||
def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":1, "params":[]}'.bytes)
|
||||
then:
|
||||
act.type == ProxyCall.RpcType.SINGLE
|
||||
act.ids.size() == 1
|
||||
act.ids[0] == 1
|
||||
act.items.size() == 1
|
||||
with(act.items[0]) {
|
||||
id == 0
|
||||
method == "net_peerCount"
|
||||
payload.toStringUtf8() == "[]"
|
||||
}
|
||||
}
|
||||
|
||||
def "Parse basic with string id"() {
|
||||
when:
|
||||
def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":"ggk19K5a", "params":[]}'.bytes)
|
||||
then:
|
||||
act.type == ProxyCall.RpcType.SINGLE
|
||||
act.ids.size() == 1
|
||||
act.ids[0] == "ggk19K5a"
|
||||
act.items.size() == 1
|
||||
with(act.items[0]) {
|
||||
id == 0
|
||||
method == "net_peerCount"
|
||||
payload.toStringUtf8() == "[]"
|
||||
}
|
||||
}
|
||||
|
||||
def "Parse basic without params"() {
|
||||
when:
|
||||
def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":2}'.bytes)
|
||||
then:
|
||||
act.type == ProxyCall.RpcType.SINGLE
|
||||
act.ids.size() == 1
|
||||
act.ids[0] == 2
|
||||
act.items.size() == 1
|
||||
with(act.items[0]) {
|
||||
id == 0
|
||||
method == "net_peerCount"
|
||||
payload.toStringUtf8() == "[]"
|
||||
}
|
||||
}
|
||||
|
||||
def "Parse with parameters"() {
|
||||
when:
|
||||
def act = reader.apply('{"jsonrpc":"2.0", "method":"net_peerCount", "id":1, "params":["0x015f"]}'.bytes)
|
||||
then:
|
||||
act.type == ProxyCall.RpcType.SINGLE
|
||||
act.ids.size() == 1
|
||||
act.ids[0] == 1
|
||||
act.items.size() == 1
|
||||
with(act.items[0]) {
|
||||
id == 0
|
||||
method == "net_peerCount"
|
||||
payload.toStringUtf8() == '["0x015f"]'
|
||||
}
|
||||
}
|
||||
|
||||
def "Parse single batch"() {
|
||||
when:
|
||||
def act = reader.apply('[{"jsonrpc":"2.0", "method":"net_peerCount", "id":1, "params":[]}]'.bytes)
|
||||
then:
|
||||
act.type == ProxyCall.RpcType.BATCH
|
||||
act.ids.size() == 1
|
||||
act.ids[0] == 1
|
||||
act.items.size() == 1
|
||||
with(act.items[0]) {
|
||||
id == 0
|
||||
method == "net_peerCount"
|
||||
payload.toStringUtf8() == "[]"
|
||||
}
|
||||
}
|
||||
|
||||
def "Parse multi batch"() {
|
||||
when:
|
||||
def act = reader.apply('[{"jsonrpc":"2.0", "method":"net_peerCount", "id":"xdd", "params":[]}, {"jsonrpc":"2.0", "method":"foo_bar", "id":4, "params":[143, false]}]'.bytes)
|
||||
then:
|
||||
act.type == ProxyCall.RpcType.BATCH
|
||||
act.ids.size() == 2
|
||||
act.ids[0] == "xdd"
|
||||
act.ids[1] == 4
|
||||
act.items.size() == 2
|
||||
with(act.items[0]) {
|
||||
id == 0
|
||||
method == "net_peerCount"
|
||||
payload.toStringUtf8() == "[]"
|
||||
}
|
||||
with(act.items[1]) {
|
||||
id == 1
|
||||
method == "foo_bar"
|
||||
payload.toStringUtf8() == '[143,false]'
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,23 +17,21 @@
|
||||
package io.emeraldpay.dshackle.rpc
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper
|
||||
import com.google.protobuf.ByteString
|
||||
import io.emeraldpay.api.proto.BlockchainOuterClass
|
||||
import io.emeraldpay.api.proto.Common
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.quorum.BroadcastQuorum
|
||||
import io.emeraldpay.dshackle.quorum.QuorumReaderFactory
|
||||
import io.emeraldpay.dshackle.quorum.QuorumRpcReader
|
||||
import io.emeraldpay.dshackle.reader.Reader
|
||||
import io.emeraldpay.dshackle.test.TestingCommons
|
||||
import io.emeraldpay.dshackle.quorum.AlwaysQuorum
|
||||
import io.emeraldpay.dshackle.quorum.NonEmptyQuorum
|
||||
import io.emeraldpay.dshackle.upstream.Multistream
|
||||
import io.emeraldpay.dshackle.upstream.Selector
|
||||
import io.emeraldpay.dshackle.upstream.MultistreamHolder
|
||||
import io.emeraldpay.dshackle.upstream.ethereum.NativeCallRouter
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
|
||||
import io.emeraldpay.grpc.Chain
|
||||
import io.infinitape.etherjar.rpc.ReactorRpcClient
|
||||
import io.infinitape.etherjar.rpc.RpcException
|
||||
import io.infinitape.etherjar.rpc.RpcResponseError
|
||||
import reactor.core.publisher.Mono
|
||||
@@ -42,7 +40,6 @@ import spock.lang.Ignore
|
||||
import spock.lang.Specification
|
||||
|
||||
import java.time.Duration
|
||||
import java.util.concurrent.TimeoutException
|
||||
|
||||
class NativeCallSpec extends Specification {
|
||||
|
||||
@@ -107,7 +104,7 @@ class NativeCallSpec extends Specification {
|
||||
1 * read(_) >> Mono.just(new QuorumRpcReader.Result("\"foo\"".bytes, 1))
|
||||
}
|
||||
}
|
||||
def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, quorum,
|
||||
def call = new NativeCall.CallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
|
||||
new NativeCall.ParsedCallDetails("eth_test", []))
|
||||
|
||||
when:
|
||||
@@ -127,7 +124,7 @@ class NativeCallSpec extends Specification {
|
||||
1 * read(_) >> Mono.empty()
|
||||
}
|
||||
}
|
||||
def call = new NativeCall.CallContext(1, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, quorum,
|
||||
def call = new NativeCall.CallContext(1, TestingCommons.multistream(TestingCommons.api()), Selector.empty, quorum,
|
||||
new NativeCall.ParsedCallDetails("eth_test", []))
|
||||
|
||||
when:
|
||||
@@ -179,7 +176,7 @@ class NativeCallSpec extends Specification {
|
||||
|
||||
when:
|
||||
def resp = nativeCall.buildResponse(
|
||||
new NativeCall.CallContext<byte[]>(1561, TestingCommons.aggregatedUpstream(TestingCommons.api()), Selector.empty, new AlwaysQuorum(), objectMapper.writeValueAsBytes(json))
|
||||
new NativeCall.CallContext<byte[]>(1561, TestingCommons.multistream(TestingCommons.api()), Selector.empty, new AlwaysQuorum(), objectMapper.writeValueAsBytes(json))
|
||||
)
|
||||
then:
|
||||
resp.id == 1561
|
||||
@@ -229,11 +226,114 @@ class NativeCallSpec extends Specification {
|
||||
def resp = nativeCall.prepareCall(req)
|
||||
then:
|
||||
StepVerifier.create(resp)
|
||||
.expectErrorMatches({t -> t instanceof NativeCall.CallFailure && t.id == 0})
|
||||
.expectErrorMatches({ t -> t instanceof NativeCall.CallFailure && t.id == 0 })
|
||||
// .expectComplete()
|
||||
.verify(Duration.ofSeconds(1))
|
||||
}
|
||||
|
||||
def "Prepare call"() {
|
||||
setup:
|
||||
def upstreams = Mock(MultistreamHolder)
|
||||
def nativeCall = new NativeCall(upstreams)
|
||||
|
||||
def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
|
||||
.setChain(Common.ChainRef.CHAIN_ETHEREUM)
|
||||
.addItems(
|
||||
BlockchainOuterClass.NativeCallItem.newBuilder()
|
||||
.setId(1)
|
||||
.setMethod("eth_test")
|
||||
.setPayload(ByteString.copyFromUtf8("[]"))
|
||||
)
|
||||
.build()
|
||||
when:
|
||||
def act = nativeCall.prepareCall(req, TestingCommons.emptyMultistream())
|
||||
.collectList().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act.size() == 1
|
||||
with(act[0]) {
|
||||
id == 1
|
||||
payload.method == "eth_test"
|
||||
payload.params == "[]"
|
||||
}
|
||||
}
|
||||
|
||||
def "Prepare call without payload"() {
|
||||
setup:
|
||||
def upstreams = Mock(MultistreamHolder)
|
||||
def nativeCall = new NativeCall(upstreams)
|
||||
|
||||
def req = BlockchainOuterClass.NativeCallRequest.newBuilder()
|
||||
.setChain(Common.ChainRef.CHAIN_ETHEREUM)
|
||||
.addItems(
|
||||
BlockchainOuterClass.NativeCallItem.newBuilder()
|
||||
.setId(1)
|
||||
.setMethod("eth_test")
|
||||
)
|
||||
.build()
|
||||
when:
|
||||
def act = nativeCall.prepareCall(req, TestingCommons.emptyMultistream())
|
||||
.collectList().block(Duration.ofSeconds(1))
|
||||
then:
|
||||
act.size() == 1
|
||||
with(act[0]) {
|
||||
id == 1
|
||||
payload.method == "eth_test"
|
||||
payload.params == ""
|
||||
}
|
||||
}
|
||||
|
||||
def "Parse empty params"() {
|
||||
setup:
|
||||
def nativeCall = new NativeCall(Stub(MultistreamHolder))
|
||||
def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
|
||||
new NativeCall.RawCallDetails("eth_test", "[]"))
|
||||
when:
|
||||
def act = nativeCall.parseParams(ctx)
|
||||
then:
|
||||
act.id == 1
|
||||
act.payload.params == []
|
||||
act.payload.method == "eth_test"
|
||||
}
|
||||
|
||||
def "Parse none params"() {
|
||||
setup:
|
||||
def nativeCall = new NativeCall(Stub(MultistreamHolder))
|
||||
def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
|
||||
new NativeCall.RawCallDetails("eth_test", ""))
|
||||
when:
|
||||
def act = nativeCall.parseParams(ctx)
|
||||
then:
|
||||
act.id == 1
|
||||
act.payload.params == []
|
||||
act.payload.method == "eth_test"
|
||||
}
|
||||
|
||||
def "Parse single param"() {
|
||||
setup:
|
||||
def nativeCall = new NativeCall(Stub(MultistreamHolder))
|
||||
def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
|
||||
new NativeCall.RawCallDetails("eth_test", "[false]"))
|
||||
when:
|
||||
def act = nativeCall.parseParams(ctx)
|
||||
then:
|
||||
act.id == 1
|
||||
act.payload.params == [false]
|
||||
act.payload.method == "eth_test"
|
||||
}
|
||||
|
||||
def "Parse multi param"() {
|
||||
setup:
|
||||
def nativeCall = new NativeCall(Stub(MultistreamHolder))
|
||||
def ctx = new NativeCall.CallContext(1, Stub(Multistream), Selector.empty, new AlwaysQuorum(),
|
||||
new NativeCall.RawCallDetails("eth_test", "[false, 123]"))
|
||||
when:
|
||||
def act = nativeCall.parseParams(ctx)
|
||||
then:
|
||||
act.id == 1
|
||||
act.payload.params == [false, 123]
|
||||
act.payload.method == "eth_test"
|
||||
}
|
||||
|
||||
@Ignore
|
||||
//TODO
|
||||
def "Calls cache before remote"() {
|
||||
@@ -241,7 +341,7 @@ class NativeCallSpec extends Specification {
|
||||
def upstreams = Stub(MultistreamHolder)
|
||||
def nativeCall = new NativeCall(upstreams)
|
||||
def api = TestingCommons.api()
|
||||
def upstream = TestingCommons.aggregatedUpstream(api)
|
||||
def upstream = TestingCommons.multistream(api)
|
||||
|
||||
def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10,
|
||||
upstream,
|
||||
@@ -259,7 +359,7 @@ class NativeCallSpec extends Specification {
|
||||
setup:
|
||||
def upstreams = Stub(MultistreamHolder)
|
||||
def nativeCall = new NativeCall(upstreams)
|
||||
def upstream = TestingCommons.aggregatedUpstream(TestingCommons.api())
|
||||
def upstream = TestingCommons.multistream(TestingCommons.api())
|
||||
|
||||
def ctx = new NativeCall.CallContext<NativeCall.ParsedCallDetails>(10,
|
||||
upstream,
|
||||
|
||||
@@ -16,15 +16,12 @@
|
||||
*/
|
||||
package io.emeraldpay.dshackle.test
|
||||
|
||||
import com.fasterxml.jackson.core.Version
|
||||
import com.fasterxml.jackson.databind.DeserializationFeature
|
||||
import com.fasterxml.jackson.databind.ObjectMapper
|
||||
import com.fasterxml.jackson.databind.module.SimpleModule
|
||||
|
||||
import io.emeraldpay.dshackle.FileResolver
|
||||
import io.emeraldpay.dshackle.Global
|
||||
import io.emeraldpay.dshackle.cache.Caches
|
||||
import io.emeraldpay.dshackle.cache.CachesFactory
|
||||
import io.emeraldpay.dshackle.config.CacheConfig
|
||||
import io.emeraldpay.dshackle.reader.EmptyReader
|
||||
import io.emeraldpay.dshackle.reader.Reader
|
||||
import io.emeraldpay.dshackle.upstream.Multistream
|
||||
import io.emeraldpay.dshackle.upstream.calls.DirectCallMethods
|
||||
@@ -33,9 +30,6 @@ import io.emeraldpay.dshackle.upstream.ethereum.EthereumUpstream
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcRequest
|
||||
import io.emeraldpay.dshackle.upstream.rpcclient.JsonRpcResponse
|
||||
import io.emeraldpay.grpc.Chain
|
||||
import io.infinitape.etherjar.rpc.JacksonRpcConverter
|
||||
|
||||
import java.text.SimpleDateFormat
|
||||
|
||||
class TestingCommons {
|
||||
|
||||
@@ -54,16 +48,20 @@ class TestingCommons {
|
||||
return new EthereumUpstreamMock(Chain.ETHEREUM, api, new DirectCallMethods(methods))
|
||||
}
|
||||
|
||||
static Multistream aggregatedUpstream(Reader<JsonRpcRequest, JsonRpcResponse> api) {
|
||||
return aggregatedUpstream(upstream(api))
|
||||
static Multistream multistream(Reader<JsonRpcRequest, JsonRpcResponse> api) {
|
||||
return multistream(upstream(api))
|
||||
}
|
||||
|
||||
static Multistream aggregatedUpstream(EthereumUpstream up) {
|
||||
static Multistream multistream(EthereumUpstream up) {
|
||||
return new EthereumMultistream(Chain.ETHEREUM, [up], Caches.default()).tap {
|
||||
start()
|
||||
}
|
||||
}
|
||||
|
||||
static Multistream emptyMultistream() {
|
||||
return multistream(new EmptyReader<JsonRpcRequest, JsonRpcResponse>())
|
||||
}
|
||||
|
||||
static CachesFactory emptyCaches() {
|
||||
return new CachesFactory(new CacheConfig())
|
||||
}
|
||||
|
||||
@@ -88,7 +88,7 @@ class EthereumReaderSpec extends Specification {
|
||||
def api = TestingCommons.api()
|
||||
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
|
||||
|
||||
def upstream = TestingCommons.aggregatedUpstream(api)
|
||||
def upstream = TestingCommons.multistream(api)
|
||||
def reader = new EthereumReader(upstream, caches, calls)
|
||||
|
||||
when:
|
||||
@@ -109,7 +109,7 @@ class EthereumReaderSpec extends Specification {
|
||||
def api = TestingCommons.api()
|
||||
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
|
||||
|
||||
def upstream = TestingCommons.aggregatedUpstream(api)
|
||||
def upstream = TestingCommons.multistream(api)
|
||||
def reader = new EthereumReader(upstream, caches, calls)
|
||||
|
||||
when:
|
||||
@@ -146,7 +146,7 @@ class EthereumReaderSpec extends Specification {
|
||||
.build()
|
||||
def api = TestingCommons.api()
|
||||
api.answer("eth_getBlockByHash", ["0xf85b826fdf98ee0f48f7db001be00472e63ceb056846f4ecac5f0c32878b8ab2", false], blockJson)
|
||||
def upstream = TestingCommons.aggregatedUpstream(api)
|
||||
def upstream = TestingCommons.multistream(api)
|
||||
def reader = new EthereumReader(upstream, caches, calls)
|
||||
|
||||
when:
|
||||
@@ -184,7 +184,7 @@ class EthereumReaderSpec extends Specification {
|
||||
|
||||
def api = TestingCommons.api()
|
||||
api.answer("eth_getTransactionByHash", [txJson.hash.toHex()], txJson)
|
||||
def upstream = TestingCommons.aggregatedUpstream(api)
|
||||
def upstream = TestingCommons.multistream(api)
|
||||
def reader = new EthereumReader(upstream, caches, calls)
|
||||
|
||||
when:
|
||||
@@ -200,7 +200,7 @@ class EthereumReaderSpec extends Specification {
|
||||
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0x10")
|
||||
api.answerOnce("eth_getBalance", ["0x70b91ff87a902b53dc6e2f6bda8bb9b330ccd30c", "latest"], "0xff")
|
||||
EthereumUpstreamMock upstream = new EthereumUpstreamMock(Chain.ETHEREUM, api)
|
||||
def upstreams = TestingCommons.aggregatedUpstream(upstream)
|
||||
def upstreams = TestingCommons.multistream(upstream)
|
||||
def reader = new EthereumReader(upstreams, Caches.default(), calls)
|
||||
reader.start()
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ class NativeCallRouterSpec extends Specification {
|
||||
def methods = new DefaultEthereumMethods(Chain.ETHEREUM)
|
||||
def router = new NativeCallRouter(
|
||||
new EthereumReader(
|
||||
TestingCommons.aggregatedUpstream(TestingCommons.api()),
|
||||
TestingCommons.multistream(TestingCommons.api()),
|
||||
Caches.default(),
|
||||
ConstantFactory.constantFactory(new DefaultEthereumMethods(Chain.ETHEREUM))
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user