link.py (6598B)
1 import logging 2 import datetime 3 import unittest 4 import os 5 import uuid 6 7 from whee.mem import MemStore 8 from usawa.store import LedgerStore 9 from usawa import Entry, Ledger, UnitIndex, DemoWallet 10 from usawa.link import EntryLink 11 12 logging.basicConfig(level=logging.DEBUG) 13 logg = logging.getLogger() 14 15 testdir = os.path.realpath(os.path.dirname(__file__)) 16 17 18 class TestLink(unittest.TestCase): 19 20 def setUp(self): 21 uidx = UnitIndex() 22 self.ledger = Ledger(uidx) 23 self.linker = EntryLink(self.ledger) 24 25 26 def test_link_dup(self): 27 o = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC)) 28 self.linker.link(o) 29 o = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 30 self.linker.link(o) 31 o = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC)) 32 with self.assertRaises(FileExistsError): 33 self.linker.link(o) 34 35 36 def test_link_explicit(self): 37 uu = uuid.uuid4() 38 o = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 39 self.linker.link(o) 40 o = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 41 self.linker.link(o, link_uuid=uu) 42 o = Entry(serial=1337, tx_date=datetime.datetime.now(datetime.UTC)) 43 self.linker.link(o, link_uuid=uu) 44 r = self.linker.get_for(o) 45 self.assertEqual(len(r), 2) 46 self.assertEqual(r[0], 42) 47 self.assertEqual(r[1], 666) 48 r = self.linker.get(o) 49 self.assertEqual(r, str(uu)) 50 51 52 def test_link_implicit(self): 53 uu = uuid.uuid4() 54 entry_a = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 55 entry_b = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 56 self.linker.link_to(entry_a, entry_b) 57 r = self.linker.get_for(entry_a) 58 self.assertEqual(len(r), 1) 59 self.assertEqual(r[0], 666) 60 r = self.linker.get_for(entry_b) 61 self.assertEqual(len(r), 1) 62 self.assertEqual(r[0], 42) 63 64 65 def test_link_implicit_ref(self): 66 linker = EntryLink(self.ledger, refs=True) 67 uu = uuid.uuid4() 68 entry_a = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 69 entry_b = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 70 linker.link_to(entry_a, entry_b) 71 o = Entry(serial=0, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 72 r = linker.get_for(o) 73 self.assertEqual(len(r), 2) 74 self.assertEqual(r[0], 42) 75 self.assertEqual(r[1], 666) 76 77 78 def test_link_entry_serialize(self): 79 uu = uuid.uuid4() 80 entry_a = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 81 entry_b = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 82 self.linker.link_to(entry_a, entry_b) 83 v = entry_a.serialize(linker=self.linker) 84 linker = EntryLink(self.ledger) 85 o = Entry.deserialize(v, linker=linker) 86 v = linker.get(o) 87 self.assertEqual(v, str(uu)) 88 89 90 91 def test_link_store(self): 92 wallet = DemoWallet() 93 uu = uuid.uuid4() 94 db = MemStore() 95 store = LedgerStore(db, self.ledger) 96 entry_a = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 97 entry_b = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 98 self.linker.link_to(entry_a, entry_b) 99 entry_a.sign(wallet) 100 store.add_entry(entry_a, linker=self.linker) 101 entry_b.sign(wallet) 102 store.add_entry(entry_b, linker=self.linker) 103 104 linker = EntryLink(self.ledger) 105 entry_a_recover = store.get_entry(entry_a, linker=linker) 106 v = linker.get(entry_a_recover) 107 self.assertEqual(v, str(uu)) 108 109 entry_b_recover = store.get_entry(entry_b, linker=linker) 110 r = linker.get_for(entry_b_recover) 111 self.assertEqual(len(r), 1) 112 self.assertEqual(r[0], 42) 113 114 r = linker.get_for(entry_a_recover) 115 self.assertEqual(len(r), 1) 116 self.assertEqual(r[0], 666) 117 118 119 def test_link_store_update(self): 120 wallet = DemoWallet() 121 uu = uuid.uuid4() 122 db = MemStore() 123 store = LedgerStore(db, self.ledger) 124 entry_a = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 125 entry_b = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 126 entry_c = Entry(serial=1337, tx_date=datetime.datetime.now(datetime.UTC)) 127 self.linker.link_to(entry_a, entry_b) 128 entry_a.sign(wallet) 129 store.add_entry(entry_a, linker=self.linker) 130 entry_b.sign(wallet) 131 store.add_entry(entry_b, linker=self.linker) 132 entry_c.sign(wallet) 133 store.add_entry(entry_c) 134 135 linker = EntryLink(self.ledger) 136 entry_a_recover = store.get_entry(entry_a, linker=linker) 137 store.get_entry(entry_b, linker=linker) 138 linker.link_to(entry_a, entry_c) 139 store.add_entry(entry_c, overwrite=True, linker=linker) 140 141 linker_updated = EntryLink(self.ledger) 142 entry_a_updated = store.get_entry(entry_a, linker=linker_updated) 143 store.get_entry(entry_b, linker=linker_updated) 144 store.get_entry(entry_c, linker=linker_updated) 145 r = linker_updated.get_for(entry_c) 146 self.assertEqual(len(r), 2) 147 self.assertEqual(r[0], 42) 148 self.assertEqual(r[1], 666) 149 150 r = linker_updated.get_for(entry_a) 151 self.assertEqual(len(r), 2) 152 self.assertEqual(r[0], 666) 153 self.assertEqual(r[1], 1337) 154 155 156 def test_link_store_index(self): 157 wallet = DemoWallet() 158 uu = uuid.uuid4() 159 db = MemStore() 160 store = LedgerStore(db, self.ledger) 161 entry_a = Entry(serial=42, tx_date=datetime.datetime.now(datetime.UTC), ref=str(uu)) 162 entry_b = Entry(serial=666, tx_date=datetime.datetime.now(datetime.UTC)) 163 entry_c = Entry(serial=1337, tx_date=datetime.datetime.now(datetime.UTC)) 164 self.linker.link_to(entry_a, entry_b) 165 entry_a.sign(wallet) 166 store.add_entry(entry_a, linker=self.linker, update_link=True) 167 entry_b.sign(wallet) 168 store.add_entry(entry_b, linker=self.linker, update_link=True) 169 entry_c.sign(wallet) 170 store.add_entry(entry_c) 171 172 link = EntryLink(self.ledger) 173 store.get_link(link, str(uu)) 174 r = link.get_for(entry_a) 175 self.assertEqual(len(r), 1) 176 r = link.get_for(entry_b) 177 self.assertEqual(len(r), 1) 178 179 180 if __name__ == '__main__': 181 unittest.main()