Merge branch 'dk-mock-data-providers' into 'develop'

Mock data providers

See merge request !336

hive/conf.py

@@ -53,6 +53,8 @@ class Conf():
         add('--exit-after-sync', help='exit when sync is completed', action='store_true')
         add('--test-profile', type=strtobool, env_var='TEST_PROFILE', help='(debug) profile execution', default=False)
         add('--log-virtual-op-calls', type=strtobool, env_var='LOG_VIRTUAL_OP_CALLS', help='(debug) log virtual op calls and responses', default=False)
+        add('--mock-block-data-path', type=str, env_var='MOCK_BLOCK_DATA_PATH', help='(debug/testing) load additional data from block data file')
+        add('--mock-vops-data-path', type=str, env_var='MOCK_VOPS_DATA_PATH', help='(debug/testing) load additional data from virtual operations data file')
 
         # logging
         add('--log-timestamp', help='Output timestamp in log', action='store_true')

hive/indexer/blocks.py

@@ -164,6 +164,8 @@ class Blocks:
             key = None
 
             op_type = vop['type']
+            if op_type not in registered_ops_stats:
+                continue
             op_value = vop['value']
             op_value['block_num'] = block_num
             key = "{}/{}".format(op_value['author'], op_value['permlink'])

hive/indexer/mock_block_provider.py

+""" Data provider for test operations """
+import logging
+import os
+from hive.indexer.mock_data_provider import MockDataProvider
+
+log = logging.getLogger(__name__)
+
+class MockBlockProvider(MockDataProvider):
+    """ Data provider for test ops """
+    @classmethod
+    def load_block_data(cls, data_path):
+        if os.path.isdir(data_path):
+            log.warning("Loading mock block data from directory: {}".format(data_path))
+            cls.add_block_data_from_directory(data_path)
+        else:
+            log.warning("Loading mock block data from file: {}".format(data_path))
+            cls.add_block_data_from_file(data_path)
+
+    @classmethod
+    def add_block_data_from_directory(cls, dir_name):
+        for name in os.listdir(dir_name):
+            file_path = os.path.join(dir_name, name)
+            if os.path.isfile(file_path) and file_path.endswith(".json"):
+                cls.add_block_data_from_file(file_path)
+
+    @classmethod
+    def add_block_data_from_file(cls, file_name):
+        from json import load
+        data = {}
+        with open(file_name, "r") as src:
+            data = load(src)
+        for block_num, transactions in data.items():
+            cls.add_block_data(block_num, transactions)
+
+    @classmethod
+    def add_block_data(cls, block_num, transactions):
+        if block_num in cls.block_data:
+            cls.block_data[str(block_num)].extend(transactions)
+        else:
+            cls.block_data[str(block_num)] = transactions
+
+    @classmethod
+    def get_block_data(cls, block_num, pop=False):
+        if pop:
+            return cls.block_data.pop(str(block_num), None)
+        return cls.block_data.get(str(block_num), None)
+
+    @classmethod
+    def get_max_block_number(cls):
+        block_numbers = [int(block) for block in cls.block_data]
+        block_numbers.append(0)
+        return max(block_numbers)
+
+    @classmethod
+    def get_blocks_greater_than(cls, block_num):
+        return sorted([int(block) for block in cls.block_data if int(block) >= block_num])

hive/indexer/mock_data_provider.py

+""" Data provider for test operations """
+from json import dumps
+
+class MockDataProvider():
+    """ Data provider for test operations """
+    block_data = {}
+
+    @classmethod
+    def print_data(cls):
+        print(dumps(cls.block_data, indent=4, sort_keys=True))

hive/indexer/mock_vops_provider.py

+""" Data provider for test vops """
+import logging
+import os
+from hive.indexer.mock_data_provider import MockDataProvider
+
+log = logging.getLogger(__name__)
+
+class MockVopsProvider(MockDataProvider):
+    """ Data provider for test vops """
+    @classmethod
+    def load_block_data(cls, data_path):
+        if os.path.isdir(data_path):
+            log.warning("Loading mock virtual ops data from directory: {}".format(data_path))
+            cls.add_block_data_from_directory(data_path)
+        else:
+            log.warning("Loading mock virtual ops data from file: {}".format(data_path))
+            cls.add_block_data_from_file(data_path)
+
+    @classmethod
+    def add_block_data_from_directory(cls, dir_name):
+        for name in os.listdir(dir_name):
+            file_path = os.path.join(dir_name, name)
+            if os.path.isfile(file_path) and file_path.endswith(".json"):
+                cls.add_block_data_from_file(file_path)
+
+    @classmethod
+    def add_block_data_from_file(cls, file_name):
+        from json import load
+        data = {}
+        with open(file_name, "r") as src:
+            data = load(src)
+        cls.add_block_data(data)
+
+    @classmethod
+    def add_block_data(cls, data):
+        if 'ops' in data:
+            if 'ops' in cls.block_data:
+                cls.block_data['ops'].extend(data['ops'])
+            else:
+                cls.block_data['ops'] = data['ops']
+
+        if 'ops_by_block' in data:
+            if 'ops_by_block' not in cls.block_data:
+                cls.block_data['ops_by_block'] = []
+
+        for ops in data['ops_by_block']:
+            for obb_ops in cls.block_data['ops_by_block']:
+                if ops['block'] == obb_ops['block']:
+                    obb_ops['ops'].extend(ops['ops'])
+
+    @classmethod
+    def get_block_data(cls, block_num):
+        ret = {}
+        if 'ops' in cls.block_data:
+            for ops in cls.block_data['ops']:
+                if ops['block'] == block_num:
+                    ret['timestamp'] = ops['timestamp']
+                    if 'ops' in ret:
+                        ret['ops'].append(ops)
+                    else:
+                        ret['ops'] = [ops]
+        if 'ops_by_block' in cls.block_data:
+            for ops in cls.block_data['ops_by_block']:
+                if ops['block'] == block_num:
+                    ret['timestamp'] = ops['timestamp']
+                    if 'ops_by_block' in ret:
+                        ret['ops_by_block'].extend(ops['ops'])
+                    else:
+                        ret['ops_by_block'] = ops['ops']
+        return ret

hive/indexer/sync.py

@@ -28,6 +28,9 @@ from hive.utils.stats import PrometheusClient as PC
 from hive.utils.stats import BroadcastObject
 from hive.utils.communities_rank import update_communities_posts_and_rank
 
+from hive.indexer.mock_block_provider import MockBlockProvider
+from hive.indexer.mock_vops_provider import MockVopsProvider
+
 from datetime import datetime
 
 log = logging.getLogger(__name__)
@@ -218,6 +221,16 @@ class Sync:
         from hive.db.schema import DB_VERSION as SCHEMA_DB_VERSION
         log.info("database_schema_version : %s", SCHEMA_DB_VERSION)
 
+        mock_block_data_path = self._conf.get("mock_block_data_path")
+        if mock_block_data_path:
+            MockBlockProvider.load_block_data(mock_block_data_path)
+            MockBlockProvider.print_data()
+
+        mock_vops_data_path = self._conf.get("mock_vops_data_path")
+        if mock_vops_data_path:
+            MockVopsProvider.load_block_data(mock_vops_data_path)
+            MockVopsProvider.print_data()
+
         # ensure db schema up to date, check app status
         DbState.initialize()
         Blocks.setup_own_db_access(self._db)

hive/steem/client.py

 """Tight and reliable steem API client for hive indexer."""
+
 import logging
 
 from time import perf_counter as perf
@@ -8,6 +9,8 @@ from hive.utils.stats import Stats
 from hive.utils.normalize import parse_amount, steem_amount, vests_amount
 from hive.steem.http_client import HttpClient
 from hive.steem.block.stream import BlockStream
+from hive.indexer.mock_block_provider import MockBlockProvider
+from hive.indexer.mock_vops_provider import MockVopsProvider
 
 logger = logging.getLogger(__name__)
 
@@ -58,10 +61,20 @@ class SteemClient:
         """
         result = self.__exec('get_block', {'block_num': num})
         if 'block' in result:
-            return result['block']
+            ret = result['block']
+            data = MockBlockProvider.get_block_data(num, True)
+            if data is not None:
+                ret["transactions"].extend(data["transactions"])
+                ret["transaction_ids"].extend(data["transaction_ids"])
+            return ret
         elif strict:
             raise Exception('block %d not available' % num)
         else:
+            # if block does not exist in hived but exist in Mock Provider
+            # return block from block provider
+            data = MockBlockProvider.get_block_data(num, True)
+            if data is not None:
+                return data
             return None
 
     def stream_blocks(self, start_from, trail_blocks=0, max_gap=100):
@@ -71,6 +84,11 @@ class SteemClient:
     def _gdgp(self):
         ret = self.__exec('get_dynamic_global_properties')
         assert 'time' in ret, "gdgp invalid resp: %s" % ret
+        mock_max_block_number = MockBlockProvider.get_max_block_number()
+        if mock_max_block_number > ret['head_block_number']:
+            ret['time'] = MockBlockProvider.get_block_data(mock_max_block_number)['timestamp']
+        ret['head_block_number'] = max([int(ret['head_block_number']), mock_max_block_number])
+        #ret['last_irreversible_block_num'] = max([int(ret['last_irreversible_block_num']), mock_max_block_number])
         return ret
 
     def head_time(self):
@@ -140,6 +158,12 @@ class SteemClient:
             num = int(block['block_id'][:8], base=16)
             blocks[num] = block
 
+        for block_num in block_nums:
+            data = MockBlockProvider.get_block_data(block_num, True)
+            if data is not None:
+                blocks[block_num]["transactions"].extend(data["transactions"])
+                blocks[block_num]["transaction_ids"].extend(data["transaction_ids"])
+
         return [blocks[x] for x in block_nums]
 
     def get_virtual_operations(self, block):
@@ -155,6 +179,21 @@ class SteemClient:
 
     def enum_virtual_ops(self, conf, begin_block, end_block):
         """ Get virtual ops for range of blocks """
+        def add_mock_vops(ret, from_block, end_block):
+            for block_num in range(from_block, end_block):
+                mock_vops = MockVopsProvider.get_block_data(block_num)
+                if mock_vops:
+                    if block_num in ret:
+                        if 'ops_by_block' in mock_vops:
+                            ret[block_num]['ops'].extend([op['op'] for op in mock_vops['ops_by_block'] if op['block'] == block_num])
+                        if 'ops' in mock_vops:
+                            ret[block_num]['ops'].extend([op['op'] for op in mock_vops['ops'] if op['block'] == block_num])
+                    else:
+                        if 'ops_by_block' in mock_vops:
+                            ret[block_num] = {'timestamp':mock_vops['timestamp'], "ops" : [op['op'] for op in mock_vops['ops_by_block'] if op['block'] == block_num]}
+                        if 'ops' in mock_vops:
+                            ret[block_num] = {'timestamp':mock_vops['timestamp'], "ops" : [op['op'] for op in mock_vops['ops'] if op['block'] == block_num]}
+
         ret = {}
 
         from_block = begin_block
@@ -199,15 +238,19 @@ class SteemClient:
             next_block = call_result['next_block_range_begin']
 
             if next_block == 0:
+                add_mock_vops(ret, from_block, end_block)
                 return ret
 
             if next_block < begin_block:
                 logger.error( "Next next block nr {} returned by enum_virtual_ops is smaller than begin block {}.".format( next_block, begin_block ) )
-                return ret;
+                add_mock_vops(ret, from_block, end_block)
+                return ret
 
             # Move to next block only if operations from current one have been processed completely.
             from_block = next_block
 
+        add_mock_vops(ret, begin_block, end_block)
+
         return ret
 
     def get_comment_pending_payouts(self, comments):

mock_data/examples/block_data/mock_block_data_example_001.json

+{
+  "5000001": {
+    "previous": "004c4b40245ffb07380a393fb2b3d841b76cdaec",
+    "timestamp": "2016-09-15T19:47:24",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [
+      {
+        "ref_block_num": 5000000,
+        "ref_block_prefix": 0,
+        "expiration": "2020-03-23T12:08:00",
+        "operations": [
+          {
+            "type": "create_claimed_account_operation",
+            "value": {
+              "creator": "esteemapp",
+              "new_account_name": "tester1",
+              "owner": {
+                "weight_threshold": 1,
+                "account_auths": [],
+                "key_auths": [
+                  [
+                    "",
+                    1
+                  ]
+                ]
+              },
+              "active": {
+                "weight_threshold": 1,
+                "account_auths": [],
+                "key_auths": [
+                  [
+                    "",
+                    1
+                  ]
+                ]
+              },
+              "posting": {
+                "weight_threshold": 1,
+                "account_auths": [],
+                "key_auths": [
+                  [
+                    "",
+                    1
+                  ]
+                ]
+              },
+              "memo_key": "",
+              "json_metadata": "",
+              "extensions": []
+            }
+          },
+          {
+            "type": "create_claimed_account_operation",
+            "value": {
+              "creator": "esteemapp",
+              "new_account_name": "tester2",
+              "owner": {
+                "weight_threshold": 1,
+                "account_auths": [],
+                "key_auths": [
+                  [
+                    "",
+                    1
+                  ]
+                ]
+              },
+              "active": {
+                "weight_threshold": 1,
+                "account_auths": [],
+                "key_auths": [
+                  [
+                    "",
+                    1
+                  ]
+                ]
+              },
+              "posting": {
+                "weight_threshold": 1,
+                "account_auths": [],
+                "key_auths": [
+                  [
+                    "",
+                    1
+                  ]
+                ]
+              },
+              "memo_key": "",
+              "json_metadata": "",
+              "extensions": []
+            }
+          }
+        ],
+        "extensions": [],
+        "signatures": [
+          ""
+        ]
+      }
+    ],
+    "block_id": "004c4b4100000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  },
+  "5000002": {
+    "previous": "004c4b4100000000000000000000000000000000",
+    "timestamp": "2016-09-15T19:47:27",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [
+      {
+        "ref_block_num": 5000001,
+        "ref_block_prefix": 1,
+        "expiration": "2020-03-23T12:17:00",
+        "operations": [
+          {
+            "type": "custom_json_operation",
+            "value": {
+              "required_auths": [],
+              "required_posting_auths": [
+                "tester1"
+              ],
+              "id": "follow",
+              "json": "[\"follow\",{\"follower\":\"tester1\",\"following\":\"tester2\",\"what\":[\"blog\"]}]"
+            }
+          }
+        ]
+      }
+    ],
+    "block_id": "004c4b4200000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  },
+  "5000003": {
+    "previous": "004c4b4200000000000000000000000000000000",
+    "timestamp": "2016-09-15T19:47:30",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [],
+    "block_id": "004c4b4300000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  },
+  "5000004": {
+    "previous": "004c4b4300000000000000000000000000000000",
+    "timestamp": "2016-09-15T19:47:33",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [],
+    "block_id": "004c4b4400000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  },
+  "5000005": {
+    "previous": "004c4b4400000000000000000000000000000000",
+    "timestamp": "2016-09-15T19:47:36",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [],
+    "block_id": "004c4b4500000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  },
+  "5000006": {
+    "previous": "004c4b4500000000000000000000000000000000",
+    "timestamp": "2016-09-15T19:47:39",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [],
+    "block_id": "004c4b4600000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  }
+}
\ No newline at end of file

mock_data/examples/block_data/mock_block_data_example_002.json

+{
+  "5000007": {
+    "previous": "004c4b4600000000000000000000000000000000",
+    "timestamp": "2016-09-15T19:47:39",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [],
+    "block_id": "004c4b4700000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  },
+  "5000008": {
+    "previous": "004c4b4700000000000000000000000000000000",
+    "timestamp": "2016-09-15T19:47:39",
+    "witness": "initminer",
+    "transaction_merkle_root": "0000000000000000000000000000000000000000",
+    "extensions": [],
+    "witness_signature": "",
+    "transactions": [],
+    "block_id": "004c4b4800000000000000000000000000000000",
+    "signing_key": "",
+    "transaction_ids": []
+  }
+}
\ No newline at end of file

mock_data/examples/vops_data/mock_vops_data_example_001.json

+{
+  "ops": [
+    {
+      "trx_id": "0000000000000000000000000000000000000000",
+      "block": 300,
+      "trx_in_block": 4294967295,
+      "op_in_trx": 0,
+      "virtual_op": 1,
+      "timestamp": "2016-03-24T16:20:30",
+      "op": {
+        "type": "producer_reward_operation",
+        "value": {
+          "producer": "tester1",
+          "vesting_shares": {
+            "amount": "1000000",
+            "precision": 6,
+            "nai": "@@000000037"
+          }
+        }
+      },
+      "operation_id": "9223372039063639274"
+    }
+  ],
+  "ops_by_block": [],
+  "next_block_range_begin": 10977,
+  "next_operation_begin": 0
+}
\ No newline at end of file

mock_data/examples/vops_data/mock_vops_data_example_002.json

+{
+  "ops": [
+    {
+      "trx_id": "0000000000000000000000000000000000000000",
+      "block": 301,
+      "trx_in_block": 4294967295,
+      "op_in_trx": 0,
+      "virtual_op": 1,
+      "timestamp": "2016-03-24T16:20:33",
+      "op": {
+        "type": "producer_reward_operation",
+        "value": {
+          "producer": "tester1",
+          "vesting_shares": {
+            "amount": "1000000",
+            "precision": 6,
+            "nai": "@@000000037"
+          }
+        }
+      },
+      "operation_id": "9223372039063639274"
+    }
+  ],
+  "ops_by_block": [],
+  "next_block_range_begin": 10977,
+  "next_operation_begin": 0
+}
\ No newline at end of file