event-msg

Simple, embedded news post vehicle for EVM smart contracts
Log | Files | Refs

test_basic.py (3966B)

      1 # standard imports
      2 import os
      3 import unittest
      4 import json
      5 import logging
      6 
      7 # external imports
      8 from chainlib.eth.unittest.ethtester import EthTesterCase
      9 from chainlib.eth.nonce import RPCNonceOracle
     10 from chainlib.eth.constant import ZERO_CONTENT
     11 from chainlib.connection import RPCConnection
     12 from chainlib.eth.tx import receipt
     13 from chainlib.eth.address import to_checksum_address
     14 from hexathon import strip_0x
     15 from hexathon import same as same_hex
     16 
     17 # local imports
     18 from eth_event_msg import EventMsg
     19 
     20 logging.basicConfig(level=logging.DEBUG)
     21 logg = logging.getLogger()
     22 
     23 
     24 hash_of_foo = '2c26b46b68ffc68ff99b453c1d30413413422d706483bfa0f98a5e886266e7ae'
     25 
     26 
     27 class Test(EthTesterCase):
     28 
     29     def setUp(self):
     30         super(Test, self).setUp()
     31         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     32         c = EventMsg(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
     33         (tx_hash, o) = c.constructor(self.accounts[0])
     34         self.conn = RPCConnection.connect(self.chain_spec, 'default')
     35         self.conn.do(o)
     36         o = receipt(tx_hash)
     37         r = self.conn.do(o)
     38         self.assertEqual(r['status'], 1)
     39         self.address = to_checksum_address(r['contract_address'])
     40         logg.debug('smart contract published with hash {}'.format(tx_hash))
     41 
     42 
     43     def test_hash(self):
     44         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     45         c = EventMsg(self.chain_spec)
     46         o = c.encode_digest(self.address, hash_of_foo, sender_address=self.accounts[0])
     47         r = self.conn.do(o)
     48         mh = c.parse_to_hash(r)
     49         self.assertEqual(mh[:4], '1220')
     50         self.assertEqual(mh[4:68], hash_of_foo)
     51 
     52 
     53     def test_uri(self):
     54         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     55         c = EventMsg(self.chain_spec)
     56         o = c.to_uri(self.address, hash_of_foo, sender_address=self.accounts[0])
     57         r = self.conn.do(o)
     58         mh = c.parse_to_uri(r)
     59         self.assertEqual(mh, 'sha256:' + hash_of_foo)
     60 
     61 
     62     def test_msg(self):
     63         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     64         c = EventMsg(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
     65 
     66         o = c.get_msg(self.address, sender_address=self.accounts[0]) 
     67         r = self.conn.do(o)
     68         mh = c.parse_to_hash(r)
     69         self.assertEqual(mh[:4], '1220')
     70         self.assertEqual(mh[4:68], strip_0x(ZERO_CONTENT))
     71 
     72         (tx_hash, o) = c.set_msg(self.address, self.accounts[0], hash_of_foo)
     73         self.conn.do(o)
     74         o = receipt(tx_hash)
     75         r = self.conn.do(o)
     76         self.assertEqual(r['status'], 1)
     77 
     78         lg = r['logs'][0]['topics'][0]
     79         logg.debug('log {}'.format(lg))
     80         self.assertTrue(same_hex(lg, '502ae868d71f78c5d099f033dd3007a4f58aebb3a2816544fe690f8b9604b56c'))
     81 
     82         o = c.get_msg(self.address, sender_address=self.accounts[0]) 
     83         r = self.conn.do(o)
     84         mh = c.parse_to_hash(r)
     85         self.assertEqual(mh[:4], '1220')
     86         self.assertEqual(mh[4:68], hash_of_foo)
     87 
     88 
     89     def test_change_codec(self):
     90         nonce_oracle = RPCNonceOracle(self.accounts[0], self.rpc)
     91         c = EventMsg(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle)
     92 
     93         (tx_hash, o) = c.set_msg(self.address, self.accounts[0], hash_of_foo)
     94         self.conn.do(o)
     95 
     96         (tx_hash, o) = c.add_codec(self.address, self.accounts[0], 32, 0xe3, 'ipfs')
     97         self.conn.do(o)
     98         o = receipt(tx_hash)
     99         r = self.conn.do(o)
    100         self.assertEqual(r['status'], 1)
    101 
    102         (tx_hash, o) = c.set_msg_codec(self.address, self.accounts[0], 0xe3)
    103         self.conn.do(o)
    104         o = receipt(tx_hash)
    105         r = self.conn.do(o)
    106         self.assertEqual(r['status'], 1)
    107       
    108         o = c.get_msg(self.address, sender_address=self.accounts[0]) 
    109         r = self.conn.do(o)
    110         mh = c.parse_to_hash(r)
    111         self.assertEqual(mh[:4], 'e320')
    112         self.assertEqual(mh[4:68], strip_0x(ZERO_CONTENT))
    113 
    114 
    115 if __name__ == '__main__':
    116     unittest.main()
    117