store.py (6032B)
1 import logging 2 import datetime 3 import unittest 4 import os 5 import copy 6 import uuid 7 8 import lxml.etree 9 from whee.mem import MemStore 10 11 from usawa import Ledger, UnitIndex, EntryPart, Entry, DemoWallet, Asset 12 from usawa.store import LedgerStore 13 from usawa.crypto import ACL 14 15 logging.basicConfig(level=logging.DEBUG) 16 logg = logging.getLogger() 17 18 testdir = os.path.realpath(os.path.dirname(__file__)) 19 20 21 class TestStore(unittest.TestCase): 22 23 def setUp(self): 24 self.store = MemStore() 25 self.parent = bytes.fromhex('0cf9180a764aba863a67b6d72f0918bc131c6772642cb2dce5a34f0a702f9470ddc2bf125c12198b1995c233c34b4afd346c54a2334c350a948a51b6e8b4e6b6') 26 self.ref = '1bda7dfa-b8fd-400d-8b42-1d2861ad7f70' 27 self.description = "foo bar baz" 28 self.dtreg = datetime.datetime.now() 29 30 31 def test_store_entry(self): 32 uidx = UnitIndex('FOO') 33 ledger = Ledger(uidx, serial=42, base=self.parent) 34 store = LedgerStore(self.store, ledger) 35 dst = EntryPart('FOO', 'asset', 'foo', 1337) 36 src = EntryPart('FOO', 'income', 'foo', 1337, debit=True) 37 o = Entry(42, datetime.datetime.strptime('2025-11-11', '%Y-%m-%d'), parent=self.parent, ref=self.ref, description=self.description, tx_datereg=self.dtreg, unitindex=uidx) 38 o.add_part(src, debit=True) 39 o.add_part(dst) 40 wallet = DemoWallet() 41 o.sign(wallet) 42 store.add_entry(o) 43 44 acl = ACL.from_wallet(wallet) 45 r = store.get_entry(o.serial, acl=acl) 46 self.assertEqual(r.ref, o.ref) 47 self.assertEqual(r.description, o.description) 48 49 50 def test_store_entry_attach(self): 51 uidx = UnitIndex('FOO') 52 ledger = Ledger(uidx, serial=42, base=self.parent) 53 store = LedgerStore(self.store, ledger) 54 dst = EntryPart('FOO', 'asset', 'foo', 1337) 55 src = EntryPart('FOO', 'income', 'foo', 1337, debit=True) 56 o = Entry(42, datetime.datetime.strptime('2025-11-11', '%Y-%m-%d'), parent=self.parent, ref=self.ref, description=self.description, tx_datereg=self.dtreg, unitindex=uidx) 57 o.add_part(src, debit=True) 58 o.add_part(dst) 59 60 fp = os.path.join(testdir, 'test.xml') 61 asset = Asset.from_file(fp, description='foobar') 62 store.add_asset(asset) 63 o.attach(asset) 64 65 wallet = DemoWallet() 66 o.sign(wallet) 67 store.add_entry(o) 68 69 acl = ACL.from_wallet(wallet) 70 r = store.get_entry(o.serial, acl=acl) 71 self.assertEqual(r.ref, o.ref) 72 self.assertEqual(r.description, o.description) 73 self.assertEqual(r.attachment[0].description, 'foobar') 74 75 76 def test_store_ledger(self): 77 uidx = UnitIndex('FOO') 78 wallet = DemoWallet() 79 acl = ACL.from_wallet(wallet) 80 ledger = Ledger(uidx, serial=41, base=self.parent, acl=acl, wallet=wallet) 81 store = LedgerStore(self.store, ledger) 82 83 dst = EntryPart('FOO', 'asset', 'foo', 1337) 84 src = EntryPart('FOO', 'income', 'foo', 1337, debit=True) 85 o = Entry(ledger.next_serial(), datetime.datetime.strptime('2025-11-11', '%Y-%m-%d'), parent=self.parent, ref=self.ref, description=self.description, tx_datereg=self.dtreg, unitindex=uidx) 86 o.add_part(src, debit=True) 87 o.add_part(dst) 88 o.sign(wallet) 89 store.add_entry(o) 90 91 ref = str(uuid.uuid4()) 92 parent = o.sum()[0] 93 description = 'barbarbar' 94 dtreg = datetime.datetime.now() 95 dst = EntryPart('FOO', 'expense', 'bar', 4200) 96 src = EntryPart('FOO', 'liability', 'bar', 4200, debit=True) 97 o = Entry(ledger.next_serial(), datetime.datetime.strptime('2025-11-12', '%Y-%m-%d'), parent=parent, ref=ref, description=description, tx_datereg=dtreg, unitindex=uidx) 98 o.add_part(src, debit=True) 99 o.add_part(dst) 100 o.sign(wallet) 101 store.add_entry(o) 102 103 ledger.sign() 104 store.load(acl=acl) 105 106 107 def test_store_import(self): 108 fp = os.path.join(testdir, 'import.xml') 109 ledger = Ledger.from_file(fp) 110 store = LedgerStore(self.store, ledger) 111 store.put_all(store_assets=True) 112 113 # TODO: less hacky test, perhaps a ledger.rewind() to get to zero state with everything else intact? 114 topic = ledger.topic 115 uidx = ledger.uidx 116 acl = ledger.acl 117 ledger = Ledger(uidx, topic=topic, acl=acl) 118 store = LedgerStore(self.store, ledger) 119 store.load(acl=acl) 120 # TODO: improve this test 121 self.assertEqual(len(ledger.entries), 2) 122 123 124 def test_store_restore(self): 125 uidx = UnitIndex('FOO') 126 wallet = DemoWallet() 127 acl = ACL.from_wallet(wallet) 128 ledger = Ledger(uidx, base=self.parent, acl=acl, wallet=wallet) 129 store = LedgerStore(self.store, ledger) 130 131 dst = EntryPart('FOO', 'asset', 'foo', 1337) 132 src = EntryPart('FOO', 'income', 'foo', 1337, debit=True) 133 o = Entry(ledger.next_serial(), datetime.datetime.strptime('2025-11-11', '%Y-%m-%d'), parent=self.parent, ref=self.ref, description=self.description, tx_datereg=self.dtreg, unitindex=uidx) 134 o.add_part(src, debit=True) 135 o.add_part(dst) 136 o.sign(wallet) 137 store.add_entry(o) 138 139 ref = str(uuid.uuid4()) 140 parent = o.sum()[0] 141 description = 'barbarbar' 142 dtreg = datetime.datetime.now() 143 dst = EntryPart('FOO', 'expense', 'bar', 4200) 144 src = EntryPart('FOO', 'liability', 'bar', 4200, debit=True) 145 o = Entry(ledger.next_serial(), datetime.datetime.strptime('2025-11-12', '%Y-%m-%d'), parent=parent, ref=ref, description=description, tx_datereg=dtreg, unitindex=uidx) 146 o.add_part(src, debit=True) 147 o.add_part(dst) 148 o.sign(wallet) 149 store.add_entry(o) 150 151 ledger.truncate() 152 s = ledger.to_string() 153 ledger = Ledger.from_string(s) 154 store = LedgerStore(self.store, ledger) 155 store.restore() 156 157 self.assertEqual(len(ledger.entries), 2) 158 159 160 if __name__ == '__main__': 161 unittest.main()