taint

Crypto forensics for private use
git clone git://git.defalsify.org/taint.git
Log | Files | Refs | LICENSE

test_cache.py (4644B)

      1 # standard imports
      2 import unittest
      3 import os
      4 import copy
      5 import logging
      6 
      7 # external imports
      8 from chainlib.chain import ChainSpec
      9 
     10 # local imports
     11 from taint.name import for_label
     12 from taint.cache import (
     13         Cache,
     14         from_index,
     15         )
     16 from taint.store.file import FileStore
     17 from taint.account import Account
     18 
     19 # test imports
     20 from tests.base import TestBase
     21 
     22 logging.basicConfig(level=logging.DEBUG)
     23 logg = logging.getLogger()
     24 
     25 
     26 class TestBasic(TestBase):
     27 
     28     def setUp(self):
     29         super(TestBasic, self).setUp()
     30 
     31         self.chain_spec = ChainSpec('foo', 'bar', 42, 'baz')
     32         self.account = os.urandom(20)
     33 
     34         self.filename = for_label(self.chain_spec, self.account, self.salt)
     35         self.filepath = os.path.join(self.session_data_dir, self.filename)
     36         self.store = FileStore(self.filepath)
     37 
     38 
     39     def test_divide(self):
     40         cache = Cache(self.chain_spec, self.bits_size, None)
     41         cache.add_subject(self.alice)
     42         cache.add_subject(self.bob)
     43         cache.add_object(self.eve)
     44         cache.add_object(self.mallory)
     45         someaccount = Account(self.chain_spec, os.urandom(20))
     46 
     47         (subjects, objects) = cache.divide([self.alice.account, self.bob.account, self.eve.account, self.mallory.account, someaccount.account])
     48 
     49         self.assertTrue(self.alice in subjects)
     50         self.assertFalse(self.alice in objects)
     51         
     52         self.assertTrue(self.bob in subjects)
     53         self.assertFalse(self.bob in objects)
     54 
     55         self.assertFalse(self.eve in subjects)
     56         self.assertTrue(self.eve in objects)
     57 
     58         self.assertFalse(self.mallory in subjects)
     59         self.assertTrue(self.mallory in objects)
     60 
     61         self.assertFalse(someaccount in subjects)
     62         self.assertFalse(someaccount in objects)
     63 
     64     
     65     def test_create_cache(self):
     66         cache = Cache(self.chain_spec, self.bits_size)
     67         cache.add_subject(self.alice)
     68         cache.add_object(self.bob)
     69 
     70         block_height = 42
     71         tx_index = 13
     72 
     73         self.assertNotEqual(self.alice.tags.get(), self.bob.tags.get())
     74 
     75         match = cache.add_tx(self.alice.account, self.bob.account, block_height, tx_index)
     76         self.assertNotEqual(match, None)
     77 
     78         self.assertEqual(self.alice.tags.get(), self.bob.tags.get())
     79 
     80 
     81     def test_state(self):
     82         cache = Cache(self.chain_spec, self.bits_size)
     83 
     84         cache.add_subject(self.alice)
     85         cache.add_subject(self.bob)
     86         cache.add_object(self.eve)
     87 
     88         first_block_height = 42
     89         first_tx_index = 13
     90         cache.add_tx(self.alice.account, self.bob.account, first_block_height, first_tx_index)
     91 
     92         new_block_height = 666
     93         new_tx_index = 1337
     94         cache.add_tx(self.alice.account, self.eve.account, new_block_height, new_tx_index)
     95 
     96         cache.first_block_height == first_block_height
     97         cache.first_tx_index == first_tx_index
     98         cache.last_block_height == new_block_height
     99         cache.last_tx_index == new_tx_index
    100 
    101 
    102     def test_recover(self):
    103         cache = Cache(self.chain_spec, self.bits_size)
    104 
    105         cache.add_subject(self.alice)
    106         cache.add_subject(self.bob)
    107         cache.add_object(self.eve)
    108 
    109         first_block_height = 42
    110         first_tx_index = 13
    111         cache.add_tx(self.alice.account, self.bob.account, first_block_height, first_tx_index)
    112 
    113         new_block_height = 666
    114         new_tx_index = 1337
    115         cache.add_tx(self.alice.account, self.eve.account, new_block_height, new_tx_index)
    116 
    117         self.assertTrue(cache.have(first_block_height, first_tx_index))
    118         self.assertTrue(cache.have(new_block_height, new_tx_index))
    119 
    120         cache = Cache(self.chain_spec, self.bits_size, cache_bloom=copy.copy(cache.cache_bloom))
    121         self.assertTrue(cache.have(first_block_height, first_tx_index))
    122         self.assertTrue(cache.have(new_block_height, new_tx_index))
    123 
    124 
    125     def test_serialize(self):
    126         cache = Cache(self.chain_spec, self.bits_size)
    127 
    128         cache.add_subject(self.alice)
    129         cache.add_subject(self.bob)
    130         cache.add_object(self.eve)
    131 
    132         first_block_height = 42
    133         first_tx_index = 13
    134         cache.add_tx(self.alice.account, self.bob.account, first_block_height, first_tx_index)
    135 
    136         new_block_height = 666
    137         new_tx_index = 1337
    138         cache.add_tx(self.alice.account, self.eve.account, new_block_height, new_tx_index)
    139 
    140         b = cache.serialize()
    141         cache_recovered = cache.from_serialized(self.chain_spec, b)
    142         self.assertEqual(cache_recovered.first_block_height, first_block_height)
    143         self.assertEqual(cache_recovered.first_tx_index, first_tx_index)
    144 
    145 
    146 if __name__ == '__main__':
    147     unittest.main()