add table tests for filter events

tests/e2e_web3js_test.go

@@ -1,9 +1,10 @@
 package tests
 
 import (
+	"testing"
+
 	"github.com/onflow/cadence"
 	"github.com/onflow/flow-emulator/emulator"
-	"testing"
 )
 
 func TestWeb3_E2E(t *testing.T) {
@@ -36,6 +37,10 @@ func TestWeb3_E2E(t *testing.T) {
 		runWeb3Test(t, "eth_streaming_test")
 	})
 
+	t.Run("streaming of entities and subscription with filters", func(t *testing.T) {
+		runWeb3Test(t, "eth_streaming_filters_test")
+	})
+
 	t.Run("batch run transactions", func(t *testing.T) {
 		runWeb3TestWithSetup(t, "eth_batch_retrieval_test", func(emu emulator.Emulator) error {
 			tx1, err := cadence.NewString("f9015880808301e8488080b901086060604052341561000f57600080fd5b60eb8061001d6000396000f300606060405260043610603f576000357c0100000000000000000000000000000000000000000000000000000000900463ffffffff168063c6888fa1146044575b600080fd5b3415604e57600080fd5b606260048080359060200190919050506078565b6040518082815260200191505060405180910390f35b60007f24abdb5865df5079dcc5ac590ff6f01d5c16edbc5fab4e195d9febd1114503da600783026040518082815260200191505060405180910390a16007820290509190505600a165627a7a7230582040383f19d9f65246752244189b02f56e8d0980ed44e7a56c0b200458caad20bb002982052fa09c05a7389284dc02b356ec7dee8a023c5efd3a9d844fa3c481882684b0640866a057e96d0a71a857ed509bb2b7333e78b2408574b8cc7f51238f25c58812662653")

tests/web3js/eth_streaming_filters_test.js

@@ -0,0 +1,107 @@
+// todo add test that uses different log filters
+// - specified address and all topics
+// - not specified address (all logs)
+// - specified address and single topic
+// - specified address and multiple topics
+// - combinations of from / to blocks
+
+
+const helpers = require("./helpers");
+const {Web3} = require("web3");
+const conf = require("./config");
+const {assert} = require("chai");
+const storageABI = require('../fixtures/storageABI.json')
+
+const timeout = 20
+
+async function assertFilterLogs(contract, filterObj, expectedLogs) {
+    const subscription = await contract.events.Calculated(filterObj)
+
+    let subId = new Promise(res => subscription.on("connected", res))
+
+    let allLogs = []
+    let logs = new Promise((res, rej) => subscription.on("data", data => {
+        allLogs.push(data)
+        console.log("### event", data.returnValues, allLogs.length, expectedLogs.length)
+        // we do this timeout as a trick, to wait if we receive more logs than we should
+        // since resolving at the expected length right away might miss another
+        // log that would unexpectedly come after.
+        setTimeout(() => {
+            if (allLogs.length === expectedLogs.length) {
+                subscription.unsubscribe()
+
+                // after we receive all logs, we make sure each received
+                // logs matches the entry in the expected log by all the values
+                for (let i = 0; i < expectedLogs.length; i++) {
+                    let expected = expectedLogs[i]
+
+                    for (const key in expected) {
+                        let expectedVal = expected[key]
+                        let actualVal = allLogs[i].returnValues[key]
+                        console.log(actualVal, expectedVal)
+                        assert.equal("####", actualVal, expectedVal)
+                    }
+                }
+
+                res(allLogs)
+            } else if (allLogs.length > expectedLogs.length) {
+                rej(allLogs)
+            }
+        }, 500)
+    }))
+
+    let id = await subId
+    assert.isString(id)
+
+    return logs
+}
+
+it('streaming of logs using filters', async() => {
+    let contractDeployment = await helpers.deployContract("storage")
+    let contractAddress = contractDeployment.receipt.contractAddress
+
+    let repeatA = 10
+    const testValues = [
+        { numA: 1, numB: 2 },
+        { numA: -1, numB: -2 },
+        { numA: repeatA, numB: 200 },
+        { numA: repeatA, numB: 300 },
+        { numA: repeatA, numB: 400 },
+    ]
+
+    let ws = new Web3("ws://127.0.0.1:8545")
+    let storageContract = new ws.eth.Contract(storageABI, contractAddress);
+
+    // wait for subscription for a bit
+    await new Promise((res, rej) => setTimeout(() => res(), 500))
+
+    let firstBlock = await ws.eth.getBlockNumber()
+
+    let allTests = [
+        //assertFilterLogs(storageContract, { fromBlock: firstBlock }, testValues),
+        assertFilterLogs(storageContract, { fromBlock: firstBlock+1n }, testValues.splice(1, 5)),
+        //assertFilterLogs(storageContract, { fromBlock: firstBlock+2n }, testValues.splice(0, 2)),
+        //assertFilterLogs(storageContract, { fromBlock: firstBlock+3n }, testValues.splice(0, 3)),
+    ]
+
+
+    // wait for subscription for a bit
+    await new Promise((res, rej) => setTimeout(() => res(), 500))
+
+    // produce events by submitting transactions
+    for (const { numA, numB } of testValues) {
+        let res = await helpers.signAndSend({
+            from: conf.eoa.address,
+            to: contractAddress,
+            data: storageContract.methods.sum(numA, numB).encodeABI(),
+            gas: 1000000,
+            gasPrice: 0
+        })
+        assert.equal(res.receipt.status, conf.successStatus)
+    }
+
+    await Promise.all(allTests)
+    await ws.eth.clearSubscriptions()
+
+    process.exit(0)
+}).timeout(timeout*1000)

tests/web3js/eth_streaming_test.js

@@ -58,7 +58,7 @@ it('streaming of logs using filters', async() => {
     let logCount = 0
     let logHashes = []
     // subscribe to events being emitted by a deployed contract and bellow transaction interactions
-    let doneLogs = new Promise(async (res, rej) => {
+    let doneAddressLogs = new Promise(async (res, rej) => {
         let subLog = await ws.eth.subscribe('logs', {
             address: contractAddress,
         })
@@ -90,7 +90,7 @@ it('streaming of logs using filters', async() => {
     }
 
     // wait for all events to be received
-    await Promise.all([doneTxs, doneBlocks, doneLogs])
+    await Promise.all([doneTxs, doneBlocks, doneAddressLogs])
 
     // check that transaction hashes we received when submitting transactions above
     // match array of transaction hashes received from events for blocks and txs