store.py (10022B)
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, AssetStore, KeyStore 13 from usawa.crypto import ACL 14 from usawa.link import EntryLink 15 16 logging.basicConfig(level=logging.DEBUG) 17 logg = logging.getLogger() 18 19 testdir = os.path.realpath(os.path.dirname(__file__)) 20 21 hash_of_foo = bytes.fromhex('f7fbba6e0636f890e56fbbf3283e524c6fa3204ae298382d624741d0dc6638326e282c41be5e4254d8820772c5518a2c5a8c0c7f7eda19594a7eb539453e1ed7') 22 uuid_for_foo = uuid.UUID('9ce0268e-4add-4874-9a00-322dd0157c97') 23 24 25 class TestStore(unittest.TestCase): 26 27 def setUp(self): 28 self.store = MemStore() 29 self.parent = bytes.fromhex('0cf9180a764aba863a67b6d72f0918bc131c6772642cb2dce5a34f0a702f9470ddc2bf125c12198b1995c233c34b4afd346c54a2334c350a948a51b6e8b4e6b6') 30 self.ref = '1bda7dfa-b8fd-400d-8b42-1d2861ad7f70' 31 self.description = "foo bar baz" 32 self.dtreg = datetime.datetime.now() 33 34 35 def test_store_entry(self): 36 uidx = UnitIndex('FOO') 37 ledger = Ledger(uidx, serial=42, base=self.parent) 38 store = LedgerStore(self.store, ledger) 39 dst = EntryPart('FOO.Asset/foo', 1337) 40 src = EntryPart('FOO.Income/foo', 1337, debit=True) 41 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) 42 o.add_part(src) 43 o.add_part(dst) 44 wallet = DemoWallet() 45 o.sign(wallet) 46 store.add_entry(o) 47 48 acl = ACL.from_wallet(wallet) 49 r = store.get_entry(o.serial, acl=acl) 50 self.assertEqual(r.ref, o.ref) 51 self.assertEqual(r.description, o.description) 52 53 54 def test_store_entry_attach(self): 55 uidx = UnitIndex('FOO') 56 ledger = Ledger(uidx, serial=42, base=self.parent) 57 store = LedgerStore(self.store, ledger) 58 dst = EntryPart('FOO.Asset/foo', 1337) 59 src = EntryPart('FOO.Income/foo', 1337, debit=True) 60 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) 61 o.add_part(src) 62 o.add_part(dst) 63 64 fp = os.path.join(testdir, 'test.xml') 65 asset = Asset.from_file(fp, description='foobar') 66 store.add_asset(asset) 67 o.attach(asset) 68 69 wallet = DemoWallet() 70 o.sign(wallet) 71 store.add_entry(o) 72 73 acl = ACL.from_wallet(wallet) 74 r = store.get_entry(o.serial, acl=acl) 75 self.assertEqual(r.ref, o.ref) 76 self.assertEqual(r.description, o.description) 77 self.assertEqual(r.attachment[0].description, 'foobar') 78 79 80 def test_store_ledger(self): 81 uidx = UnitIndex('FOO') 82 wallet = DemoWallet() 83 acl = ACL.from_wallet(wallet) 84 ledger = Ledger(uidx, serial=41, base=self.parent, acl=acl, wallet=wallet) 85 store = LedgerStore(self.store, ledger) 86 87 dst = EntryPart('FOO.Asset/foo', 1337) 88 src = EntryPart('FOO.Income/foo', 1337, debit=True) 89 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) 90 o.add_part(src) 91 o.add_part(dst) 92 o.sign(wallet) 93 store.add_entry(o) 94 95 ref = str(uuid.uuid4()) 96 parent = o.sum()[0] 97 description = 'barbarbar' 98 dtreg = datetime.datetime.now() 99 dst = EntryPart('FOO.Expense/bar', 4200) 100 src = EntryPart('FOO.Liability/bar', 4200, debit=True) 101 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) 102 o.add_part(src) 103 o.add_part(dst) 104 o.sign(wallet) 105 store.add_entry(o) 106 107 ledger.sign() 108 store.load(acl=acl) 109 110 111 @unittest.skip('import.xml must be updated to current structure') 112 def test_store_import(self): 113 fp = os.path.join(testdir, 'import.xml') 114 ledger = Ledger.from_file(fp) 115 store = LedgerStore(self.store, ledger) 116 store.put_all(store_assets=True) 117 118 # TODO: less hacky test, perhaps a ledger.rewind() to get to zero state with everything else intact? 119 topic = ledger.topic 120 uidx = ledger.uidx 121 acl = ledger.acl 122 ledger = Ledger(uidx, topic=topic, acl=acl) 123 store = LedgerStore(self.store, ledger) 124 store.load(acl=acl) 125 # TODO: improve this test 126 self.assertEqual(len(ledger.entries), 2) 127 128 129 def test_store_restore(self): 130 uidx = UnitIndex('FOO') 131 wallet = DemoWallet() 132 acl = ACL.from_wallet(wallet) 133 ledger = Ledger(uidx, base=self.parent, acl=acl, wallet=wallet) 134 store = LedgerStore(self.store, ledger) 135 136 dst = EntryPart('FOO.Asset/foo', 1337) 137 src = EntryPart('FOO.Income/foo', 1337, debit=True) 138 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) 139 o.add_part(src) 140 o.add_part(dst) 141 o.sign(wallet) 142 store.add_entry(o) 143 144 ref = str(uuid.uuid4()) 145 parent = o.sum()[0] 146 description = 'barbarbar' 147 dtreg = datetime.datetime.now() 148 dst = EntryPart('FOO.Expense/bar', 4200) 149 src = EntryPart('FOO.Liability/bar', 4200, debit=True) 150 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) 151 o.add_part(src) 152 o.add_part(dst) 153 o.sign(wallet) 154 store.add_entry(o) 155 156 ledger.truncate() 157 s = ledger.to_string() 158 ledger = Ledger.from_string(s) 159 store = LedgerStore(self.store, ledger) 160 store.restore() 161 162 self.assertEqual(len(ledger.entries), 2) 163 164 165 166 def test_store_entry_base(self): 167 uidx = UnitIndex('FOO') 168 ledger = Ledger(uidx, serial=42, base=self.parent) 169 store = LedgerStore(self.store, ledger) 170 dst = EntryPart('FOO.Asset/foo', 1337) 171 src = EntryPart('FOO.Income/foo', 1337, debit=True) 172 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) 173 o.add_part(src) 174 o.add_part(dst) 175 o.add_tag('foo') 176 o.add_pair('bar', 'baz') 177 store.put_draft(o) 178 o = store.get_draft(o) 179 180 181 182 def test_store_asset_index(self): 183 store = AssetStore(self.store) 184 asset = Asset(digest=hash_of_foo, ref=str(uuid_for_foo)) 185 store.add_asset(asset) 186 187 asset = Asset(digest=hash_of_foo) 188 o = store.get_asset(asset) 189 self.assertEqual(o.get_digest(binary=True), hash_of_foo) 190 self.assertEqual(o.get_ref(binary=True), uuid_for_foo.bytes) 191 192 asset = Asset(ref=str(uuid_for_foo)) 193 o = store.get_asset_indexed(asset) 194 self.assertEqual(o.get_digest(binary=True), hash_of_foo) 195 self.assertEqual(o.get_ref(binary=True), uuid_for_foo.bytes) 196 197 198 def test_store_more(self): 199 uidx = UnitIndex('FOO') 200 ledger = Ledger(uidx, topic=b'foobar') 201 store = LedgerStore(self.store, ledger) 202 wallet = DemoWallet() 203 204 dst = EntryPart('FOO.Asset/foo', 1337) 205 src = EntryPart('FOO.Income/foo', 1337, debit=True) 206 o = Entry(1, datetime.datetime.strptime('2025-11-11', '%Y-%m-%d'), ref=str(uuid.uuid4()), description='foo', tx_datereg=self.dtreg, unitindex=uidx) 207 o.add_part(src) 208 o.add_part(dst) 209 210 fp = os.path.join(testdir, 'test.xml') 211 asset = Asset.from_file(fp, description='foobar') 212 store.add_asset(asset) 213 o.attach(asset) 214 o.sign(wallet) 215 store.add_entry(o, update_ledger=True) 216 217 dst = EntryPart('FOO.Expense/bar', 42) 218 src = EntryPart('FOO.Asset/bar', -42, debit=True) 219 now = datetime.datetime.now(datetime.UTC) 220 o = Entry(2, now, parent=ledger.cur, ref=str(uuid.uuid4()), description='bar', unitindex=uidx) 221 o.add_part(src) 222 o.add_part(dst) 223 o.sign(wallet) 224 store.add_entry(o) 225 226 acl = ACL.from_wallet(wallet) 227 ledger = Ledger(uidx, topic=b'foobar') 228 store = LedgerStore(self.store, ledger) 229 store.load() 230 231 232 def test_store_link(self): 233 uidx = UnitIndex('FOO') 234 ledger = Ledger(uidx, topic=b'foobar') 235 store = LedgerStore(self.store, ledger) 236 ref_a = uuid.uuid4() 237 ref_b = uuid.uuid4() 238 link = EntryLink(ledger) 239 entry_a = Entry.empty(serial=1, ref=str(ref_a)) 240 entry_b = Entry.empty(serial=2, ref=str(ref_b)) 241 link.link_to(entry_a, entry_b) 242 store.put_link(link) 243 244 link = EntryLink(ledger) 245 store.get_link(link, str(ref_a)) 246 r = link.get_for(entry_a) 247 self.assertEqual(len(r), 1) 248 r = link.get_for(entry_b) 249 self.assertEqual(len(r), 1) 250 251 252 def test_store_keys(self): 253 store = KeyStore(self.store) 254 wallet_one = DemoWallet() 255 store.add_key(wallet_one, passphrase='foo') 256 wallet_two = DemoWallet() 257 store.add_key(wallet_two, default=True, passphrase='bar') 258 r = store.get_key(DemoWallet, pubkey=wallet_one.pubkey(), passphrase='foo') 259 self.assertEqual(wallet_one.privkey(), r.privkey()) 260 r = store.get_default_key(DemoWallet) 261 v = r.pubkey() 262 self.assertEqual(wallet_two.pubkey(), v) 263 r = store.get_key(DemoWallet, pubkey=v, passphrase='bar') 264 self.assertEqual(wallet_two.privkey(), r.privkey()) 265 266 267 if __name__ == '__main__': 268 unittest.main()