add integration test

integration_tests/configs/default.jsonnet

@@ -2,7 +2,7 @@
   dotenv: '../../scripts/.env',
   'cronos_777-1': {
     cmd: 'cronosd',
-    'start-flags': '--trace',
+    'start-flags': '--trace --streamers file',
     config: {
       mempool: {
         version: 'v1',

integration_tests/network.py

@@ -55,9 +55,10 @@ def node_rpc(self, i):
         return "tcp://127.0.0.1:%d" % ports.rpc_port(self.base_port(i))
 
     def cosmos_cli(self, i=0):
-        return CosmosCLI(
-            self.base_dir / f"node{i}", self.node_rpc(i), self.chain_binary
-        )
+        return CosmosCLI(self.node_home(i), self.node_rpc(i), self.chain_binary)
+
+    def node_home(self, i=0):
+        return self.base_dir / f"node{i}"
 
     def use_websocket(self, use=True):
         self._w3 = None

integration_tests/pyproject.toml

@@ -26,6 +26,7 @@ toml = "^0.10.2"
 pysha3 = "^1.0.2"
 jsonnet = "^0.18.0"
 eth-account = { git = "https://github.com/mmsqe/eth-account.git", branch = "v0.5.8-rc0" }
+cprotobuf = { git = "https://github.com/yihuang/cprotobuf.git" }
 
 [tool.poetry.dev-dependencies]
 

integration_tests/test_streamer.py

@@ -0,0 +1,62 @@
+from cprotobuf import Field, ProtoEntity, decode_primitive
+from hexbytes import HexBytes
+
+from .utils import ADDRS
+
+
+class StoreKVPairs(ProtoEntity):
+    # the store key for the KVStore this pair originates from
+    store_key = Field("string", 1)
+    # true indicates a set operation, false indicates a delete operation
+    delete = Field("bool", 2)
+    key = Field("bytes", 3)
+    value = Field("bytes", 4)
+
+
+def decode_stream_file(data, body_cls=StoreKVPairs, header_cls=None, footer_cls=None):
+    """
+    header, body*, footer
+    """
+    header = footer = None
+    body = []
+    offset = 0
+    size, n = decode_primitive(data, "uint64")
+    offset += n
+
+    # header
+    if header_cls is not None:
+        header = header_cls()
+        header.ParseFromString(data[offset : offset + size])
+    offset += size
+
+    while True:
+        size, n = decode_primitive(data[offset:], "uint64")
+        offset += n
+        if offset + size == len(data):
+            # footer
+            if footer_cls is not None:
+                footer = footer_cls()
+                footer.ParseFromString(data[offset : offset + size])
+            offset += size
+            break
+        else:
+            # body
+            if body_cls is not None:
+                item = body_cls()
+                item.ParseFromString(data[offset : offset + size])
+                body.append(item)
+            offset += size
+    return header, body, footer
+
+
+def test_streamers(cronos):
+    """
+    - check the streaming files are created
+    - try to parse the state change sets
+    """
+    # inspect the first state change of the first tx in genesis
+    path = cronos.node_home(0) / "data/file_streamer/block-0-tx-0"
+    _, body, _ = decode_stream_file(open(path, "rb").read())
+    # creation of the validator account
+    assert body[0]["store_key"] == "acc"
+    assert body[0]["key"] == b"\x01" + HexBytes(ADDRS["validator"])

nix/testenv.nix

@@ -14,5 +14,11 @@ pkgs.poetry2nix.mkPoetryEnv {
         nativeBuildInputs = (old.nativeBuildInputs or [ ]) ++ [ self.poetry ];
       }
     );
+
+    cprotobuf = super.cprotobuf.overridePythonAttrs (
+      old: {
+        nativeBuildInputs = (old.nativeBuildInputs or [ ]) ++ [ self.cython ];
+      }
+    );
   });
 }