eth-monitor

Monitor and cache ethereum transactions with match filters
git clone git://git.defalsify.org/eth-monitor.git
Info | Log | Files | Refs | README | LICENSE

test_base.py (5120B)

      1 # standard imports
      2 import logging
      3 import unittest
      4 import os
      5 
      6 # local imports
      7 from eth_monitor.rules import *
      8 
      9 logging.basicConfig(level=logging.DEBUG)
     10 logg = logging.getLogger()
     11 
     12 
     13 class TestRule(unittest.TestCase):
     14 
     15     def setUp(self):
     16         self.alice = os.urandom(20).hex()
     17         self.bob = os.urandom(20).hex()
     18         self.carol = os.urandom(20).hex()
     19         self.dave = os.urandom(20).hex()
     20         self.x = os.urandom(20).hex()
     21         self.y = os.urandom(20).hex()
     22         self.hsh = os.urandom(32).hex()
     23 
     24     
     25     def test_address_include(self):
     26         data = b''
     27         outs = [self.alice]
     28         ins = []
     29         execs = []
     30         rule = RuleSimple(outs, ins, execs)
     31         c = AddressRules()
     32         c.include(rule)
     33         r = c.apply_rules_addresses(self.alice, self.bob, data, self.hsh)
     34         self.assertTrue(r)
     35         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
     36         self.assertFalse(r)
     37 
     38         outs = []
     39         ins = [self.alice]
     40         execs = []
     41         rule = RuleSimple(outs, ins, execs)
     42         c = AddressRules()
     43         c.include(rule)
     44         r = c.apply_rules_addresses(self.alice, self.bob, data, self.hsh)
     45         self.assertFalse(r)
     46         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
     47         self.assertTrue(r)
     48 
     49         outs = []
     50         ins = []
     51         execs = [self.x]
     52         rule = RuleSimple(outs, ins, execs)
     53         c = AddressRules()
     54         c.include(rule)
     55         r = c.apply_rules_addresses(self.alice, self.x, data, self.hsh)
     56         self.assertTrue(r)
     57         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
     58         self.assertFalse(r)
     59 
     60         data = b'deadbeef0123456789'
     61         data_match = [data[:8]]
     62         rule = RuleMethod(data_match)
     63         c = AddressRules()
     64         c.include(rule)
     65         r = c.apply_rules_addresses(self.alice, self.x, data, self.hsh)
     66         self.assertTrue(r)
     67         r = c.apply_rules_addresses(self.bob, self.alice, b'abcd' + data, self.hsh)
     68         self.assertFalse(r)
     69 
     70         rule = RuleData(data_match)
     71         c = AddressRules()
     72         c.include(rule)
     73         r = c.apply_rules_addresses(self.alice, self.x, data, self.hsh)
     74         self.assertTrue(r)
     75         r = c.apply_rules_addresses(self.bob, self.alice, b'abcd' + data, self.hsh)
     76         self.assertTrue(r)
     77 
     78 
     79     def test_address_exclude(self):
     80         data = b''
     81         outs = [self.alice]
     82         ins = []
     83         execs = []
     84         rule = RuleSimple(outs, ins, execs)
     85 
     86         c = AddressRules()
     87         c.exclude(rule)
     88         r = c.apply_rules_addresses(self.alice, self.bob, data, self.hsh)
     89         self.assertFalse(r)
     90         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
     91         self.assertFalse(r)
     92 
     93         c = AddressRules(include_by_default=True)
     94         c.exclude(rule)
     95         r = c.apply_rules_addresses(self.alice, self.bob, data, self.hsh)
     96         self.assertFalse(r)
     97         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
     98         self.assertTrue(r)
     99 
    100         outs = []
    101         ins = [self.alice]
    102         execs = []
    103         rule = RuleSimple(outs, ins, execs)
    104         c = AddressRules(include_by_default=True)
    105         c.exclude(rule)
    106         r = c.apply_rules_addresses(self.alice, self.bob, data, self.hsh)
    107         self.assertTrue(r)
    108         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
    109         self.assertFalse(r)
    110 
    111         outs = []
    112         ins = []
    113         execs = [self.x]
    114         rule = RuleSimple(outs, ins, execs)
    115         c = AddressRules(include_by_default=True)
    116         c.exclude(rule)
    117         r = c.apply_rules_addresses(self.alice, self.x, data, self.hsh)
    118         self.assertFalse(r)
    119         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
    120         self.assertTrue(r)
    121 
    122         data = b'deadbeef0123456789'
    123         data_match = [data[:8]]
    124         rule = RuleMethod(data_match)
    125         c = AddressRules(include_by_default=True)
    126         c.exclude(rule)
    127         r = c.apply_rules_addresses(self.alice, self.x, data, self.hsh)
    128         self.assertFalse(r)
    129         r = c.apply_rules_addresses(self.bob, self.alice, b'abcd' + data, self.hsh)
    130         self.assertTrue(r)
    131 
    132         rule = RuleData(data_match)
    133         c = AddressRules(include_by_default=True)
    134         c.exclude(rule)
    135         r = c.apply_rules_addresses(self.alice, self.x, data, self.hsh)
    136         self.assertFalse(r)
    137         r = c.apply_rules_addresses(self.bob, self.alice, b'abcd' + data, self.hsh)
    138         self.assertFalse(r)
    139         r = c.apply_rules_addresses(self.bob, self.alice, b'abcd', self.hsh)
    140         self.assertTrue(r)
    141 
    142     
    143     def test_address_include_exclude(self):
    144         data = b''
    145         outs = [self.alice]
    146         ins = []
    147         execs = []
    148         rule = RuleSimple(outs, ins, execs)
    149         c = AddressRules()
    150         c.include(rule)
    151         c.exclude(rule)
    152         r = c.apply_rules_addresses(self.alice, self.bob, data, self.hsh)
    153         self.assertFalse(r)
    154         r = c.apply_rules_addresses(self.bob, self.alice, data, self.hsh)
    155         self.assertFalse(r)
    156 
    157 
    158 
    159 if __name__ == '__main__':
    160     unittest.main()