""" Vrismcost — the test suite. A tax tool does not crash when it is wrong. It returns a number, confidently, and the number is wrong. That is the failure mode these tests exist to catch: every one of them pins a behaviour that would otherwise break silently. Nothing here touches the network. The fetcher is exercised through its pure functions and the engine through synthetic transactions, so the suite runs in a few seconds, offline, and gives the same answer next year as it does today. Tests that call a live chain are tests whose result depends on what strangers traded this morning. Several of these were written after the bug, not before it, and say so. python test_vrismcost.py python test_vrismcost.py -v """ from __future__ import annotations import json import tempfile import unittest from decimal import Decimal as D from pathlib import Path from fetcher import Leg, Tx, to_topic, from_topic, TRANSFER_TOPIC from engine import Engine, money, qty, rate, units import report as rep from vault import (Vault, VaultError, WrongPassphrase, format_recovery, parse_recovery, validate_passphrase) # ---------------------------------------------------------------- fixtures ME = "0x1111111111111111111111111111111111111111" COLD = "0x2222222222222222222222222222222222222222" POOL = "0x3333333333333333333333333333333333333333" ROUTER = "0x4444444444444444444444444444444444444444" FEE = "0x5555555555555555555555555555555555555555" STRANGER = "0x6666666666666666666666666666666666666666" ZERO = "0x" + "0" * 40 USDG = "0xaaaa000000000000000000000000000000000001" NVDA = "0xaaaa000000000000000000000000000000000002" TSLA = "0xaaaa000000000000000000000000000000000003" TOKENS = { USDG: {"symbol": "USDG", "decimals": 6, "name": "Global Dollar", "resolved": True}, NVDA: {"symbol": "NVDA", "decimals": 18, "name": "NVIDIA", "resolved": True}, TSLA: {"symbol": "TSLA", "decimals": 18, "name": "Tesla", "resolved": True}, } DAY = 864_000 # blocks per day at 0.1s def leg(token: str, frm: str, to: str, amount: str, i: int = 0) -> Leg: decimals = TOKENS[token]["decimals"] return Leg(token=token, frm=frm, to=to, raw=int(D(amount) * (D(10) ** decimals)), log_index=i) def tx(hash_: str, block: int, legs: list[Leg], *, native: str = "0", gas: str = "0.0001", sender: str = ME, raw_legs: int | None = None) -> Tx: return Tx(tx=hash_, block=block, timestamp=1_782_000_000 + block // 10, sender=sender, to=POOL, native_value=int(D(native) * (D(10) ** 18)), gas_wei=int(D(gas) * (D(10) ** 18)), status="0x1", raw_legs=raw_legs if raw_legs is not None else len(legs), legs=legs) def engine(**kw) -> Engine: opts = dict(wallets=[ME, COLD], tokens=TOKENS, quote_symbol="USDG") opts.update(kw) return Engine(**opts) def buy(hash_: str, block: int, paid: str, got: str, **kw) -> Tx: return tx(hash_, block, [leg(USDG, ME, POOL, paid, 0), leg(NVDA, POOL, ME, got, 1)], **kw) # ---------------------------------------------------------------- helpers class Helpers(unittest.TestCase): def test_topic_padding_round_trips(self): self.assertEqual(to_topic(ME), "0x" + "0" * 24 + ME[2:]) self.assertEqual(from_topic(to_topic(ME)), ME) def test_topic_is_case_insensitive(self): self.assertEqual(to_topic(ME.upper()), to_topic(ME.lower())) def test_transfer_topic_is_the_known_signature(self): # keccak("Transfer(address,address,uint256)"). Hardcoding it is fine; # silently mistyping it is not, and every scan depends on it. self.assertEqual( TRANSFER_TOPIC, "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") def test_units_respects_token_decimals(self): # USDG has 6 decimals, not 18. Getting this wrong moves every USDG # figure by a factor of a trillion. self.assertEqual(units(1_000_000, 6), D("1")) self.assertEqual(units(10 ** 18, 18), D("1")) def test_quantity_never_renders_in_scientific_notation(self): # normalize() alone turns 70 into 7E+1, which is not a quantity anyone # can paste into a tax return. self.assertEqual(qty(D("70")), "70") self.assertEqual(qty(D("0.00000001")), "0.00000001") def test_rate_keeps_precision_small_prices_actually_have(self): # A memecoin at four millionths of a dollar renders as 0.00 at cents # precision, which reads as broken. self.assertEqual(rate(D("0")), "0") self.assertNotEqual(rate(D("0.0000048")), "0.00") self.assertEqual(money(D("2815.504805192172")), "2815.50") # ---------------------------------------------------------------- classify class Classification(unittest.TestCase): def test_quote_out_asset_in_is_a_buy(self): e = engine() ev = e.classify(buy("0xa", 100, "5296", "40")) self.assertEqual(ev.kind, "BUY") self.assertEqual(ev.coverage, "full") self.assertFalse(ev.taxable) def test_asset_out_quote_in_is_a_sale(self): e = engine() ev = e.classify(tx("0xb", 200, [leg(NVDA, ME, POOL, "70", 0), leg(USDG, POOL, ME, "11781", 1)])) self.assertEqual(ev.kind, "SELL") self.assertTrue(ev.taxable) def test_both_sides_mine_is_a_move_not_a_disposal(self): # The single most expensive misread available: booking a wallet move as # a sale invents a gain out of nothing. e = engine() ev = e.classify(tx("0xc", 300, [leg(NVDA, ME, COLD, "30")])) self.assertEqual(ev.kind, "INTERNAL") self.assertFalse(ev.taxable) self.assertEqual(ev.gained, {}) self.assertEqual(ev.given, {}) def test_router_legs_net_away(self): # A real trade here averages 5.2 legs. Reading them individually # misreads the majority of traffic, so the net is what gets classified. e = engine() ev = e.classify(tx("0xd", 400, [ leg(USDG, ME, ROUTER, "2074.80", 0), leg(USDG, ROUTER, POOL, "2069.61", 1), leg(USDG, ROUTER, FEE, "5.19", 2), leg(NVDA, POOL, ROUTER, "5", 3), leg(NVDA, ROUTER, ME, "5", 4), ])) self.assertEqual(ev.kind, "BUY") self.assertEqual(ev.given, {"USDG": D("2074.80")}) self.assertEqual(ev.gained, {"NVDA": D("5")}) self.assertEqual(ev.raw_legs, 5) def test_native_eth_comes_from_tx_value(self): # ETH emits no Transfer log at all. One transaction in five moves some. e = engine(native_price=D("3000")) ev = e.classify(tx("0xe", 500, [leg(NVDA, POOL, ME, "2")], native="0.5")) self.assertEqual(ev.kind, "BUY") self.assertEqual(ev.given, {"ETH": D("0.5")}) def test_eth_leg_without_a_rate_is_partly_priced_not_guessed(self): e = engine(native_price=None) ev = e.classify(tx("0xf", 600, [leg(NVDA, POOL, ME, "2")], native="0.5")) self.assertEqual(ev.kind, "SWAP") self.assertEqual(ev.coverage, "partial") self.assertIn("ETH", ev.gap) def test_token_to_token_has_no_stated_value(self): e = engine() ev = e.classify(tx("0x10", 700, [leg(NVDA, ME, POOL, "10", 0), leg(TSLA, POOL, ME, "5", 1)])) self.assertEqual(ev.kind, "SWAP") self.assertEqual(ev.coverage, "none") def test_arrival_with_nothing_paid_is_not_guessed(self): e = engine() ev = e.classify(tx("0x11", 800, [leg(NVDA, STRANGER, ME, "12")])) self.assertEqual(ev.kind, "RECEIVE") self.assertIsNone(ev.taxable) self.assertEqual(ev.coverage, "none") def test_departure_with_nothing_received_is_not_guessed(self): e = engine() ev = e.classify(tx("0x12", 900, [leg(NVDA, ME, STRANGER, "12")])) self.assertEqual(ev.kind, "SEND") self.assertIsNone(ev.taxable) def test_untouched_transaction_is_unrelated(self): e = engine() ev = e.classify(tx("0x13", 1000, [leg(NVDA, STRANGER, POOL, "5")], sender=STRANGER)) self.assertEqual(ev.kind, "UNRELATED") # ---------------------------------------------------------------- corporate class CorporateActions(unittest.TestCase): def split(self, block: int, before: str, after: str) -> Tx: return tx("0xsplit", block, [leg(NVDA, ME, ZERO, before, 0), leg(NVDA, ZERO, ME, after, 1)]) def test_burn_and_mint_of_one_token_is_a_split(self): # This chain runs corporate actions as burn-and-mint, which to a # log reader is a total disposal plus a free repurchase. e = engine() ev = e.classify(self.split(1000, "60", "240")) self.assertEqual(ev.kind, "SPLIT") self.assertFalse(ev.taxable) self.assertEqual(ev.corp["ratio"], "4") def test_burn_and_mint_of_different_tokens_is_a_migration(self): e = engine() ev = e.classify(tx("0xm", 1100, [leg(NVDA, ME, ZERO, "10", 0), leg(TSLA, ZERO, ME, "10", 1)])) self.assertEqual(ev.kind, "MIGRATION") self.assertEqual(ev.corp["into"], "TSLA") def test_mint_with_nothing_paid_is_income_not_a_lot(self): e = engine() ev = e.classify(tx("0xd1", 1200, [leg(USDG, ZERO, ME, "84.60")])) self.assertEqual(ev.kind, "DISTRIBUTION") self.assertTrue(ev.taxable) def test_a_quote_asset_mint_asks_whether_it_was_income(self): # The amount of a USDG mint is stated exactly; what it was is not. A # dividend, a bridge deposit and an issuance against the owner's own # dollars all log a Transfer from 0x0, and the quote is recognised by a # symbol any token can claim. So it stays open, asked the right question. ev = engine().classify(tx("0xd1", 1200, [leg(USDG, ZERO, ME, "84.60")])) self.assertEqual((ev.kind, ev.coverage), ("DISTRIBUTION", "none")) self.assertEqual(ev.gap, "not confirmed as income") self.assertEqual(ev.gained, {"USDG": D("84.60")}) def test_a_quote_asset_mint_in_two_legs_states_the_whole_amount(self): # Only the first minted leg was read. Harmless while nothing used the # amount; wrong the moment an answer takes it as the value. ev = engine().classify(tx("0xd2", 1200, [leg(USDG, ZERO, ME, "50", 0), leg(USDG, ZERO, ME, "34.60", 1)])) self.assertEqual(ev.gained, {"USDG": D("84.60")}) def test_a_mint_of_any_other_token_still_needs_a_value(self): ev = engine().classify(tx("0xd3", 1200, [leg(NVDA, ZERO, ME, "2")])) self.assertEqual((ev.kind, ev.coverage, ev.gap), ("DISTRIBUTION", "none", "distribution has no stated value")) def test_split_preserves_total_cost_and_the_acquisition_date(self): # The invariant a split must pass. If total cost moves, the adjustment # was wrong; if the date moves, a holding period was reset on a # position nobody sold. e = engine() r = e.run([buy("0xa", 1000, "5296", "40"), self.split(2000, "40", "160")]) lot = r["lots"][0] self.assertEqual(lot["qty"], "160") self.assertEqual(money(D(lot["cost"])), "5296.00") self.assertTrue(lot["split"]) self.assertEqual(lot["acquired"], r["events"][0].at.date().isoformat()) def test_a_split_between_two_sales_lands_once(self): # Written after the bug. The split was applied again at every sale and # once more at the end, and before the sales that came first: one # 4-for-1 split turned 40 shares into 2,560. r = engine().run([buy("0xa", 1 * DAY, "5296", "40", gas="0"), tx("0xs1", 10 * DAY, [leg(NVDA, ME, POOL, "10", 0), leg(USDG, POOL, ME, "2000", 1)], gas="0"), self.split(20 * DAY, "30", "120"), tx("0xs2", 30 * DAY, [leg(NVDA, ME, POOL, "20", 0), leg(USDG, POOL, ME, "1200", 1)], gas="0")]) first, second = (d["methods"]["FIFO"] for d in r["disposals"]) self.assertEqual(first["costBasis"], "1324.00") # 10 at 132.40, before the split self.assertEqual(second["costBasis"], "662.00") # 20 at 33.10, after it lot = r["lots"][0] self.assertEqual((lot["qty"], lot["remaining"]), ("160", "100")) self.assertEqual(len(lot["split"]), 1) self.assertEqual(len(r["corporateActions"]), 1) def test_a_split_leaves_lots_bought_after_it_alone(self): r = engine().run([buy("0xa", 1 * DAY, "1000", "10", gas="0"), self.split(5 * DAY, "10", "40"), buy("0xb", 9 * DAY, "1000", "20", gas="0")]) before, after = r["lots"] self.assertEqual(before["qty"], "40") self.assertEqual(after["qty"], "20") self.assertFalse(after["split"]) # ---------------------------------------------------------------- lots class Lots(unittest.TestCase): def test_acquisition_gas_joins_the_basis(self): e = engine(native_price=D("2000")) r = e.run([buy("0xa", 100, "1000", "10", gas="0.001")]) # 0.001 ETH at 2000 = 2.00 on top of 1000 paid. self.assertEqual(money(D(r["lots"][0]["cost"])), "1002.00") def test_nothing_unpriced_ever_becomes_a_zero_basis_lot(self): # A zero-basis lot exports as 100% taxable gain. Queueing is the only # honest alternative. e = engine() r = e.run([tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")])]) self.assertEqual(r["lots"], []) self.assertEqual(len(r["queue"]), 1) def test_a_wallet_move_does_not_close_the_lot(self): e = engine() r = e.run([buy("0xa", 100, "1000", "10"), tx("0xmv", 200, [leg(NVDA, ME, COLD, "10")])]) self.assertEqual(len(r["lots"]), 1) self.assertEqual(r["lots"][0]["remaining"], "10") self.assertEqual(r["realised"], "0.00") def test_gas_counts_only_when_your_wallet_sent_the_transaction(self): # Written after the bug. A bundler's transaction, or a stranger's, had # its whole gas charged to whichever wallet it touched. e = engine(native_price=D("2000")) r = e.run([buy("0xmine", 100, "1000", "10", gas="0.001"), buy("0xrelayed", 200, "1000", "10", gas="0.01", sender=STRANGER)]) self.assertEqual([money(D(l["cost"])) for l in r["lots"]], ["1002.00", "1000.00"]) self.assertEqual(r["gasPaidByOthers"], 1) # ---------------------------------------------------------------- matching class Matching(unittest.TestCase): def history(self): """Four buys at different prices, then one sale of 70.""" return [ buy("0xb1", 1 * DAY, "5296.00", "40", gas="0"), buy("0xb2", 20 * DAY, "4156.25", "35", gas="0"), buy("0xb3", 40 * DAY, "3852.50", "25", gas="0"), buy("0xb4", 55 * DAY, "4248.00", "30", gas="0"), tx("0xsell", 60 * DAY, [leg(NVDA, ME, POOL, "70", 0), leg(USDG, POOL, ME, "11781.00", 1)], gas="0"), ] def setUp(self): self.r = engine().run(self.history()) self.m = self.r["disposals"][0]["methods"] def test_fifo_takes_the_oldest_lots_first(self): legs = self.m["FIFO"]["legs"] self.assertEqual([l["qty"] for l in legs], ["40", "30"]) self.assertEqual(self.m["FIFO"]["costBasis"], "8858.50") def test_lifo_takes_the_newest_lots_first(self): self.assertEqual(self.m["LIFO"]["legs"][0]["qty"], "30") self.assertEqual(self.m["LIFO"]["costBasis"], "9881.75") def test_hifo_takes_the_dearest_lots_first(self): legs = self.m["HIFO"]["legs"] costs = [D(l["unitCost"]) for l in legs] self.assertEqual(costs, sorted(costs, reverse=True)) self.assertEqual(self.m["HIFO"]["costBasis"], "10086.50") def test_average_pools_every_open_lot(self): self.assertEqual(len(self.m["AVERAGE"]["legs"]), 1) self.assertEqual(self.m["AVERAGE"]["legs"][0]["acquired"], "pooled") def test_every_method_sells_exactly_seventy(self): for name, m in self.m.items(): with self.subTest(method=name): total = sum(D(l["qty"]) for l in m["legs"]) self.assertEqual(total, D("70"), name) self.assertEqual(D(m["unmatched"]), D("0"), name) def test_the_method_chosen_changes_the_tax_owed(self): # The whole argument for offering four of them. gains = {k: D(v["gain"]) for k, v in self.m.items()} self.assertEqual(max(gains.values()) - min(gains.values()), D("1228.00")) self.assertLess(gains["HIFO"], gains["FIFO"]) def test_gain_is_always_proceeds_minus_basis(self): for name, m in self.m.items(): with self.subTest(method=name): self.assertEqual( D(m["gain"]), D(m["netProceeds"]) - D(m["costBasis"]), name) def test_holding_days_are_measured_to_the_disposal(self): legs = self.m["FIFO"]["legs"] self.assertGreater(legs[0]["heldDays"], legs[1]["heldDays"]) def test_a_sale_cannot_reach_a_lot_bought_after_it(self): # Written after the bug: every lot was open to every sale, including # lots that did not exist yet when the sale happened. r = engine().run([buy("0xb1", 1 * DAY, "1000", "10", gas="0"), tx("0xs", 5 * DAY, [leg(NVDA, ME, POOL, "15", 0), leg(USDG, POOL, ME, "3000", 1)], gas="0"), buy("0xb2", 9 * DAY, "5000", "10", gas="0")]) fifo = r["disposals"][0]["methods"]["FIFO"] self.assertEqual([(l["lot"], l["qty"]) for l in fifo["legs"]], [(1, "10")]) self.assertEqual(fifo["unmatched"], "5") def test_every_method_sells_from_its_own_inventory(self): # Written after the bug. A lot sold down to nothing came back for the # next sale, and LIFO, HIFO and average all sold from FIFO's leftovers. r = engine().run([buy("0xc1", 1 * DAY, "100", "1", gas="0"), buy("0xc2", 2 * DAY, "300", "1", gas="0"), tx("0xt1", 3 * DAY, [leg(NVDA, ME, POOL, "1", 0), leg(USDG, POOL, ME, "400", 1)], gas="0"), tx("0xt2", 4 * DAY, [leg(NVDA, ME, POOL, "1", 0), leg(USDG, POOL, ME, "400", 1)], gas="0")]) bases = {m: [d["methods"][m]["costBasis"] for d in r["disposals"]] for m in ("FIFO", "LIFO", "HIFO", "AVERAGE")} self.assertEqual(bases, {"FIFO": ["100.00", "300.00"], "LIFO": ["300.00", "100.00"], "HIFO": ["300.00", "100.00"], "AVERAGE": ["200.00", "200.00"]}) # ---------------------------------------------------------------- answers class Answers(unittest.TestCase): def test_an_answer_becomes_a_lot(self): got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")]) e = engine(answers={"0xr": {"action": "offchain", "value": "2568.72"}}) r = e.run([got]) self.assertEqual(len(r["lots"]), 1) self.assertEqual(money(D(r["lots"][0]["cost"])), "2568.72") def test_one_value_across_two_assets_is_divided_not_duplicated(self): # Found in testing, not in review: five answers of 125.00 produced ten # lots of 125.00 each, doubling the cost basis, because the arriving # transaction carried two tokens. got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "10", 0), leg(TSLA, STRANGER, ME, "5", 1)]) e = engine(answers={"0xr": {"action": "own", "value": "125.00"}}) r = e.run([got]) self.assertEqual(len(r["lots"]), 2) total = sum(D(l["cost"]) for l in r["lots"]) self.assertEqual(money(total), "125.00") self.assertTrue(all(l["allocation"] for l in r["lots"])) def test_a_sole_asset_is_not_marked_as_split(self): got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "10")]) e = engine(answers={"0xr": {"action": "own", "value": "100.00"}}) self.assertIsNone(e.run([got])["lots"][0]["allocation"]) def test_excluding_a_transaction_creates_nothing(self): got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "10")]) e = engine(answers={"0xr": {"action": "exclude"}}) self.assertEqual(e.run([got])["lots"], []) def test_answers_are_counted_apart_from_chain_evidence(self): txs = [buy("0xa", 100, "1000", "10"), tx("0xr", 200, [leg(NVDA, STRANGER, ME, "5")])] plain = engine().run(txs)["coverage"] answered = engine(answers={"0xr": {"action": "own", "value": "50"}} ).run(txs)["coverage"] self.assertEqual(plain["full"], 1) self.assertEqual(plain["blocked"], 1) self.assertEqual(answered["full"], 1) # the chain proved no more self.assertEqual(answered["answered"], 1) # the owner supplied one self.assertEqual(answered["blocked"], 0) def test_an_unanswered_quote_mint_holds_the_export_open(self): r = engine().run([tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")]) self.assertEqual(r["coverage"]["blocked"], 1) pairs = {x["field"]: x["value"] for x in rep.summary_rows(r, {})} self.assertTrue(pairs["status"].startswith("INCOMPLETE")) self.assertEqual([(x["value"], x["source"]) for x in rep.income_rows(r)], [(rep.UNPRICED, "UNRESOLVED")]) def test_an_open_distribution_is_income_business_not_a_disposal(self): # Written after the bug: the unanswered mint was written to the capital # gains schedule as well as the income sheet. r = engine().run([tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")]) self.assertEqual(rep.disposal_rows(r, "FIFO"), []) def test_income_on_a_quote_mint_reaches_the_income_sheet_at_its_amount(self): # Written after the bug. An `income` answer took the mint out of the # queue, the income sheet read only the queue, and the export called # itself COMPLETE with the income nowhere in it. r = engine(answers={"0xd1": {"action": "income"}}).run( [tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")]) self.assertEqual([(x["asset"], x["quantity"], x["value"], x["source"], x["tx"]) for x in rep.income_rows(r)], [("USDG", "84.6", "84.60", "user", "0xd1")]) self.assertEqual((r["coverage"]["answered"], r["coverage"]["blocked"]), (1, 0)) self.assertEqual(r["lots"], []) def test_income_stated_on_a_receive_reaches_the_income_sheet_and_the_lot(self): # The same leak for any token: the lot was built, the income never written. r = engine(answers={"0xr": {"action": "income", "value": "2747.40"}}).run( [tx("0xr", DAY, [leg(NVDA, STRANGER, ME, "12")], gas="0")]) self.assertEqual(money(D(r["lots"][0]["cost"])), "2747.40") self.assertEqual([(x["asset"], x["quantity"], x["value"], x["source"]) for x in rep.income_rows(r)], [("NVDA", "12", "2747.40", "user")]) def test_own_or_exclude_on_a_quote_mint_is_not_income(self): for action in ("own", "exclude"): with self.subTest(action): r = engine(answers={"0xd1": {"action": action}}).run( [tx("0xd1", DAY, [leg(USDG, ZERO, ME, "84.60")], gas="0")]) self.assertEqual(rep.income_rows(r), []) self.assertEqual(r["coverage"]["blocked"], 0) # ---------------------------------------------------------------- export class Export(unittest.TestCase): def setUp(self): self.r = engine().run([ buy("0xb1", 1 * DAY, "1000.00", "10", gas="0"), tx("0xsell", 30 * DAY, [leg(NVDA, ME, POOL, "10", 0), leg(USDG, POOL, ME, "1500.00", 1)], gas="0"), tx("0xr", 40 * DAY, [leg(TSLA, STRANGER, ME, "5")]), ]) def test_money_is_quantised_and_quantities_are_not(self): rows = rep.disposal_rows(self.r, "FIFO") self.assertEqual(rows[0]["cost_basis"], "1000.00") self.assertEqual(rows[0]["gain"], "500.00") self.assertNotIn("E+", rows[0]["quantity"]) def test_both_transaction_hashes_are_on_the_row(self): # One hash proves half a claim. A schedule needs the acquisition and # the disposal side by side to be audit evidence. row = rep.disposal_rows(self.r, "FIFO")[0] self.assertEqual(row["acquired_tx"], "0xb1") self.assertEqual(row["disposed_tx"], "0xsell") def test_unresolved_rows_are_exported_not_dropped(self): rows = rep.disposal_rows(self.r, "FIFO") unresolved = [r for r in rows if r["source"] == "UNRESOLVED"] self.assertEqual(len(unresolved), 1) self.assertEqual(unresolved[0]["cost_basis"], rep.UNPRICED) def test_the_summary_realised_gain_follows_the_chosen_method(self): # Written after the bug. The summary sheet printed the FIFO gain under # whatever method it named, so an export labelled HIFO disagreed with # its own disposal sheet the moment two lots had different costs. r = engine().run([ buy("0xcheap", 1 * DAY, "100.00", "1", gas="0"), buy("0xdear", 2 * DAY, "300.00", "1", gas="0"), tx("0xsell1", 30 * DAY, [leg(NVDA, ME, POOL, "1", 0), leg(USDG, POOL, ME, "400.00", 1)], gas="0"), ]) for method, expected in (("FIFO", "300.00"), ("HIFO", "100.00"), ("LIFO", "100.00")): pairs = {row["field"]: row["value"] for row in rep.summary_rows(r, {"method": method})} sheet_total = sum(D(row["gain"]) for row in rep.disposal_rows(r, method)) self.assertEqual(pairs["realised_gain"], expected, method) self.assertEqual(D(pairs["realised_gain"]), sheet_total, method) def test_the_summary_says_when_the_file_is_incomplete(self): pairs = {r["field"]: r["value"] for r in rep.summary_rows(self.r, {})} self.assertTrue(pairs["status"].startswith("INCOMPLETE")) self.assertEqual(pairs["unresolved"], "1") def test_a_clean_history_reports_complete(self): clean = engine().run([buy("0xb1", DAY, "1000", "10", gas="0")]) pairs = {r["field"]: r["value"] for r in rep.summary_rows(clean, {})} self.assertEqual(pairs["status"], "COMPLETE") def test_open_lots_carry_their_basis_forward(self): rows = rep.open_lot_rows(engine().run([buy("0xb", DAY, "1000", "10", gas="0")])) self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["cost_basis"], "1000.00") def test_open_lots_follow_the_chosen_method(self): r = engine().run([buy("0xcheap", 1 * DAY, "100.00", "1", gas="0"), buy("0xdear", 2 * DAY, "300.00", "1", gas="0"), tx("0xsell", 3 * DAY, [leg(NVDA, ME, POOL, "1", 0), leg(USDG, POOL, ME, "400.00", 1)], gas="0")]) left = {m: [row["cost_basis"] for row in rep.open_lot_rows(r, m)] for m in ("FIFO", "LIFO", "HIFO")} self.assertEqual(left, {"FIFO": ["300.00"], "LIFO": ["100.00"], "HIFO": ["100.00"]}) def test_every_sheet_writes_its_header_even_when_empty(self): with tempfile.TemporaryDirectory() as d: written = rep.write_report(str(Path(d) / "t"), self.r, {}, "FIFO") self.assertEqual(len(written), 4) for path, _ in written: first = path.read_text(encoding="utf-8").splitlines()[0] self.assertTrue(first and "," in first, path.name) self.assertFalse(first.startswith("#"), "a comment line offsets every column") # ---------------------------------------------------------------- vault class VaultTests(unittest.TestCase): def setUp(self): self.dir = tempfile.TemporaryDirectory() self.path = Path(self.dir.name) / "vault.json" def tearDown(self): self.dir.cleanup() def test_contents_round_trip_through_a_lock(self): v, _ = Vault.create("a-long-enough-passphrase", self.path) v.add_wallet(ME, "main") v.answer("0xabc", "own", "100.00") v.save() v.lock() again = Vault.unlock("a-long-enough-passphrase", self.path) self.assertEqual(again.wallets(), [ME]) self.assertEqual(again.answers()["0xabc"]["value"], "100.00") def test_nothing_readable_survives_on_disk(self): v, _ = Vault.create("a-long-enough-passphrase", self.path) v.add_wallet(ME, "my-main-wallet") v.answer("0xabc", "own", "1234.56", "bought on an exchange") v.save() raw = self.path.read_text(encoding="utf-8") for secret in (ME[2:10], "1234.56", "exchange", "my-main-wallet"): self.assertNotIn(secret, raw) def test_the_wrong_passphrase_is_refused(self): Vault.create("a-long-enough-passphrase", self.path) with self.assertRaises(WrongPassphrase): Vault.unlock("not-the-passphrase", self.path) def test_a_tampered_file_fails_its_integrity_check(self): v, _ = Vault.create("a-long-enough-passphrase", self.path) v.answer("0xabc", "own", "100") v.save() env = json.loads(self.path.read_text(encoding="utf-8")) ct = env["body"]["ciphertext"] env["body"]["ciphertext"] = ("B" if ct[0] != "B" else "C") + ct[1:] self.path.write_text(json.dumps(env), encoding="utf-8") with self.assertRaises(VaultError): Vault.unlock("a-long-enough-passphrase", self.path) def test_recovery_restores_and_retires_the_code_it_used(self): v, code = Vault.create("a-long-enough-passphrase", self.path) v.add_wallet(ME, "main") v.save() v.lock() back, fresh = Vault.recover(code, "a-different-passphrase", self.path) self.assertEqual(back.wallets(), [ME]) self.assertNotEqual(code, fresh) with self.assertRaises(VaultError): Vault.recover(code, "yet-another-passphrase", self.path) def test_recovery_codes_round_trip(self): secret = bytes(range(32)) self.assertEqual(parse_recovery(format_recovery(secret)), secret) self.assertEqual(parse_recovery(format_recovery(secret).lower()), secret) def test_short_passphrases_are_refused(self): # A vault backup can be attacked offline for as long as someone likes. with self.assertRaises(VaultError): validate_passphrase("short") # ---------------------------------------------------------------- the wire # Everything below runs the real Fetcher and the real pipeline against a # JSON-RPC node that lives in memory. Only _send is fake: the one method that # knows HTTP exists. import subprocess import sys from unittest import mock import fetcher as fm import price import vrismcost from fetcher import Fetcher, HttpStatus, RpcError, TransportError WETH = price.WETH HEAD = 3_000_000 BASE_TS = 1_782_000_000 def _word(n: int) -> str: return n.to_bytes(32, "big").hex() def _abi_string(text: str) -> str: raw = text.encode() return "0x" + _word(32) + _word(len(raw)) + raw.ljust((len(raw) + 31) // 32 * 32, b"\0").hex() class FakeNode: """Enough of a Robinhood Chain node to run a scan end to end, offline.""" def __init__(self, tokens: dict): self.tokens = tokens self.logs: list[dict] = [] self.txs: dict[str, dict] = {} self.receipts: dict[str, dict] = {} self.requests: list[str] = [] def transfer(self, tx_hash, block, index, token, frm, to, amount): decimals = self.tokens[token]["decimals"] raw = int(D(amount) * (D(10) ** decimals)) log = {"address": token, "blockNumber": hex(block), "transactionHash": tx_hash, "logIndex": hex(index), "data": "0x" + _word(raw), "topics": [TRANSFER_TOPIC, to_topic(frm), to_topic(to)]} self.logs.append(log) self.txs[tx_hash] = {"from": ME, "to": POOL, "value": "0x0"} rc = self.receipts.setdefault(tx_hash, {"gasUsed": "0x0", "effectiveGasPrice": "0x0", "status": "0x1", "logs": []}) rc["logs"].append(log) def answer(self, method, params): self.requests.append(method) if method == "eth_blockNumber": return hex(HEAD) if method == "eth_getLogs": flt = params[0] lo, hi = int(flt["fromBlock"], 16), int(flt["toBlock"], 16) out = [] for log in self.logs: if not lo <= int(log["blockNumber"], 16) <= hi: continue if flt.get("address") and log["address"] != flt["address"].lower(): continue wanted = flt.get("topics") or [] if all(w is None or log["topics"][i] in (w if isinstance(w, list) else [w]) for i, w in enumerate(wanted)): out.append(log) return out if method == "eth_getBlockByNumber": return {"timestamp": hex(BASE_TS + int(params[0], 16) // 10)} if method == "eth_getTransactionByHash": return self.txs.get(params[0]) if method == "eth_getTransactionReceipt": return self.receipts.get(params[0]) if method == "eth_call": meta = self.tokens[params[0]["to"]] return {fm.SEL_SYMBOL: _abi_string(meta["symbol"]), fm.SEL_DECIMALS: "0x" + _word(meta["decimals"]), fm.SEL_NAME: _abi_string(meta["name"])}[params[0]["data"]] raise AssertionError(f"the fake node was asked for {method}") def send(self, data: bytes) -> bytes: payload = json.loads(data) rows = payload if isinstance(payload, list) else [payload] out = [{"jsonrpc": "2.0", "id": r["id"], "result": self.answer(r["method"], r["params"])} for r in rows] return json.dumps(out if isinstance(payload, list) else out[0]).encode() def trade_node() -> FakeNode: """Buy 5 NVDA for 1,000 USDG, later sell 2 for 500.""" node = FakeNode(TOKENS) node.transfer("0x" + "a1" * 32, 1_000_000, 0, USDG, ME, POOL, "1000") node.transfer("0x" + "a1" * 32, 1_000_000, 1, NVDA, POOL, ME, "5") node.transfer("0x" + "b2" * 32, 2_000_000, 0, NVDA, ME, POOL, "2") node.transfer("0x" + "b2" * 32, 2_000_000, 1, USDG, POOL, ME, "500") return node class NodeFetcher(Fetcher): """The real Fetcher; only the wire and the clock are replaced.""" def __init__(self, node: FakeNode, **kw): super().__init__(**kw) self.node = node self.slept: list[float] = [] def _send(self, data: bytes) -> bytes: return self.node.send(data) def _sleep(self, seconds: float) -> None: self.slept.append(seconds) class ScriptedFetcher(Fetcher): """A Fetcher whose wire replays a script of outcomes, one per attempt.""" def __init__(self, script): super().__init__() self.script = list(script) self.slept: list[float] = [] def _send(self, data: bytes) -> bytes: step = self.script.pop(0) if isinstance(step, Exception): raise step return step def _sleep(self, seconds: float) -> None: self.slept.append(seconds) OK_BODY = json.dumps([{"jsonrpc": "2.0", "id": 0, "result": "0x10"}]).encode() class Transport(unittest.TestCase): def test_rate_limit_is_retried_with_a_widening_pause(self): f = ScriptedFetcher([HttpStatus(429), HttpStatus(429), OK_BODY]) self.assertEqual(f.call("eth_blockNumber", []), "0x10") self.assertEqual(f.retries, 2) self.assertEqual(f.slept, [fm.RETRY_BASE, fm.RETRY_BASE * 2]) def test_retry_after_longer_than_the_backoff_is_honoured(self): f = ScriptedFetcher([HttpStatus(429, retry_after=10.0), OK_BODY]) f.call("eth_blockNumber", []) self.assertEqual(f.slept, [10.0]) def test_a_client_error_is_not_retried(self): f = ScriptedFetcher([HttpStatus(400)]) with self.assertRaises(RpcError) as caught: f.call("eth_blockNumber", []) self.assertEqual(caught.exception.code, 400) self.assertEqual(f.slept, []) def test_a_dead_network_gives_up_after_the_retry_budget(self): f = ScriptedFetcher([TransportError("URLError")] * fm.RETRY_MAX) with self.assertRaises(RpcError) as caught: f.call("eth_blockNumber", []) self.assertIn("transport", str(caught.exception)) self.assertEqual(len(f.slept), fm.RETRY_MAX - 1) def test_every_log_query_is_paced_but_the_first(self): # Written after the bug: the oracle's log queries went out back to # back, and a full scan died of HTTP 429 at the pricing stage. empty = json.dumps([{"jsonrpc": "2.0", "id": 0, "result": []}]).encode() f = ScriptedFetcher([empty, empty, empty]) for _ in range(3): f.paced_logs({"fromBlock": "0x1", "toBlock": "0x2"}) self.assertEqual(f.slept, [fm.PACE_SECONDS, fm.PACE_SECONDS]) class Oracle(unittest.TestCase): def _swaps(self, rates): """A node holding one WETH->USDG swap per rate, all at one block.""" tokens = {price.WETH: {"symbol": "WETH", "decimals": 18, "name": "WETH"}, price.USDG: {"symbol": "USDG", "decimals": 6, "name": "USDG"}} node = FakeNode(tokens) for n, r in enumerate(rates): h = "0x" + f"{n + 1:02x}" * 32 node.transfer(h, 1_000_000, 0, price.WETH, ME, POOL, "1") node.transfer(h, 1_000_000, 1, price.USDG, POOL, ME, str(r)) return node def test_swaps_that_agree_are_a_price(self): got = price.derive_native_price(NodeFetcher(self._swaps([1880, 1881, 1882])), 1_000_000) self.assertIsNotNone(got) self.assertEqual(round(got["rate"]), 1881) def test_swaps_that_disagree_are_not_a_price(self): # Measured on chain, not invented: router legs matched as though they # were pools priced ETH at 1,412 USDG with a 76% spread, where the # neighbouring windows said about 1,890. The page promised that legs # stay unpriced when the swaps disagree; the code used the number. node = self._swaps([800, 940, 1880]) self.assertIsNone(price.derive_native_price(NodeFetcher(node), 1_000_000)) oracle = price.PriceOracle(NodeFetcher(node), bucket=100_000) self.assertIsNone(oracle.rate_at(1_000_000)) self.assertEqual(oracle.summary()["unresolvedReasons"], ["no agreeing pools"]) def test_a_rate_limited_window_is_an_unpriced_gap_not_a_dead_scan(self): # Written after the bug. The transfers were already read; losing them # all because one price window was refused is the wrong trade. f = ScriptedFetcher([HttpStatus(429)] * fm.RETRY_MAX) oracle = price.PriceOracle(f, bucket=100_000) self.assertIsNone(oracle.rate_at(1_234_567)) s = oracle.summary() self.assertEqual((s["buckets"], s["resolved"], s["unresolved"]), (1, 0, 1)) self.assertTrue(any("node unavailable" in r for r in s["unresolvedReasons"])) def test_a_quiet_window_says_so(self): node = FakeNode(TOKENS) oracle = price.PriceOracle(NodeFetcher(node), bucket=100_000) self.assertIsNone(oracle.rate_at(1_234_567)) self.assertEqual(oracle.summary()["unresolvedReasons"], ["no agreeing pools"]) class Pipeline(unittest.TestCase): def test_scanning_does_not_need_the_vault(self): # The vault pulls in scrypt and AES-GCM. Reading a wallet needs neither, # and should work where they are not installed. code = "import sys; sys.modules['vault'] = None; import vrismcost" r = subprocess.run([sys.executable, "-c", code], cwd=Path(__file__).parent, capture_output=True, text=True) self.assertEqual(r.returncode, 0, r.stderr) def test_an_empty_range_returns_no_report(self): result = vrismcost.run([ME], start=fm.LAUNCH_BLOCK, end=HEAD, fetcher=NodeFetcher(FakeNode(TOKENS))) self.assertIsNone(result["report"]) self.assertEqual(result["head"], HEAD) def test_the_pipeline_reads_a_real_shaped_chain(self): result = vrismcost.run([ME], start=fm.LAUNCH_BLOCK, end=HEAD, fetcher=NodeFetcher(trade_node())) report = result["report"] self.assertEqual(result["quote"], "USDG") self.assertEqual(report["shapes"].get("BUY"), 1) self.assertEqual(report["shapes"].get("SELL"), 1) fifo = report["disposals"][0]["methods"]["FIFO"] self.assertEqual(D(fifo["costBasis"]), D("400")) # 2 of 5 units bought for 1,000 self.assertEqual(D(fifo["gain"]), D("100")) # ---------------------------------------------------------------- 0.2.6 # Written after the pre-release audit, before the fixes. Each of these was a # way the export could be wrong while its summary said COMPLETE. def swap(hash_: str, block: int, give: str, give_amt: str, get: str, get_amt: str, **kw) -> Tx: return tx(hash_, block, [leg(give, ME, POOL, give_amt, 0), leg(get, POOL, ME, get_amt, 1)], **kw) def summary(report: dict, meta: dict | None = None) -> dict: return {row["field"]: row["value"] for row in rep.summary_rows(report, meta or {})} class Resolutions(unittest.TestCase): """The queue offered answers the engine then ignored.""" def history(self): return [buy("0xb", 1 * DAY, "1000", "10", gas="0"), swap("0xsw", 5 * DAY, NVDA, "4", TSLA, "2", gas="0")] def test_an_unanswered_swap_still_gives_up_the_lots_that_left(self): r = engine().run(self.history()) self.assertEqual(r["lots"][0]["remaining"], "6") d = r["disposals"][0] self.assertIsNone(d["proceeds"]) self.assertEqual(d["methods"]["FIFO"]["costBasis"], "400.00") self.assertIsNone(d["methods"]["FIFO"]["gain"]) self.assertTrue(summary(r)["status"].startswith("INCOMPLETE")) row = [x for x in rep.disposal_rows(r, "FIFO") if x["disposed_tx"] == "0xsw"][0] self.assertEqual((row["cost_basis"], row["proceeds"], row["gain"], row["source"]), ("400.00", rep.UNPRICED, rep.UNPRICED, "UNRESOLVED")) def test_a_value_on_a_swap_is_proceeds_out_and_basis_in(self): r = engine(answers={"0xsw": {"action": "manual", "value": "900"}}).run(self.history()) d = r["disposals"][0] fifo = d["methods"]["FIFO"] self.assertEqual((d["proceeds"], fifo["costBasis"], fifo["gain"]), ("900.00", "400.00", "500.00")) tsla = [l for l in r["lots"] if l["symbol"] == "TSLA"][0] self.assertEqual((money(D(tsla["cost"])), tsla["source"]), ("900.00", "user")) self.assertEqual(r["coverage"]["blocked"], 0) row = [x for x in rep.disposal_rows(r, "FIFO") if x["disposed_tx"] == "0xsw"][0] self.assertEqual((row["gain"], row["source"]), ("500.00", "user")) self.assertEqual(summary(r)["status"], "COMPLETE") def test_sold_off_chain_is_a_disposal_at_the_stated_proceeds(self): txs = [buy("0xb", 1 * DAY, "1000", "10", gas="0"), tx("0xout", 5 * DAY, [leg(NVDA, ME, STRANGER, "10")], gas="0")] r = engine(answers={"0xout": {"action": "sold", "value": "1500"}}).run(txs) self.assertEqual(r["disposals"][0]["methods"]["FIFO"]["gain"], "500.00") self.assertEqual(r["lots"][0]["remaining"], "0") self.assertEqual(r["coverage"]["blocked"], 0) self.assertEqual(rep.open_lot_rows(r), []) def test_an_eth_rate_answer_prices_that_transaction(self): paid = tx("0xeth", 5 * DAY, [leg(NVDA, POOL, ME, "2")], native="0.5", gas="0.001") r = engine(answers={"0xeth": {"action": "eth_price", "value": "2000"}}).run([paid]) self.assertEqual(len(r["lots"]), 1) self.assertEqual(money(D(r["lots"][0]["cost"])), "1002.00") # 0.5 ETH and 0.001 ETH gas at 2,000 self.assertEqual(r["coverage"]["blocked"], 0) def test_an_answer_that_cannot_apply_leaves_the_transaction_open(self): r = engine(answers={"0xsw": {"action": "eth_price", "value": "2000"}}).run(self.history()) self.assertEqual((r["coverage"]["answered"], r["coverage"]["blocked"]), (0, 1)) got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")]) r = engine(answers={"0xr": {"action": "offchain"}}).run([got]) self.assertEqual((r["lots"], r["coverage"]["blocked"]), ([], 1)) def test_an_exclusion_is_listed_in_the_export(self): got = tx("0xr", 100, [leg(NVDA, STRANGER, ME, "12")]) r = engine(answers={"0xr": {"action": "exclude"}}).run([got]) self.assertEqual([x["tx"] for x in r["excluded"]], ["0xr"]) pairs = summary(r) self.assertEqual((pairs["excluded_by_owner"], pairs["excluded_transactions"]), ("1", "0xr")) self.assertEqual(pairs["status"], "COMPLETE") def test_your_own_wallet_can_carry_its_original_date(self): got = tx("0xr", 40 * DAY, [leg(NVDA, STRANGER, ME, "5")]) r = engine(answers={"0xr": {"action": "own", "value": "400", "date": "2026-07-01"}}).run([got]) lot = r["lots"][0] self.assertEqual((lot["acquired"], lot["source"]), ("2026-07-01", "user")) self.assertEqual(rep.open_lot_rows(r)[0]["source"], "user") class Provenance(unittest.TestCase): """Every figure says where it came from: the chain, a rate derived from it, or you.""" def test_chain_evidence_is_marked_chain(self): r = engine().run([buy("0xb", DAY, "1000", "10", gas="0"), tx("0xs", 2 * DAY, [leg(NVDA, ME, POOL, "10", 0), leg(USDG, POOL, ME, "1500", 1)], gas="0")]) self.assertEqual(rep.disposal_rows(r, "FIFO")[0]["source"], "chain") def test_a_derived_eth_rate_is_marked_derived(self): class Rate: def rate_at(self, block): return D("2000") paid = tx("0xeth", 5 * DAY, [leg(NVDA, POOL, ME, "2")], native="0.5", gas="0") sold = tx("0xs", 9 * DAY, [leg(NVDA, ME, POOL, "2", 0), leg(USDG, POOL, ME, "1500", 1)], gas="0") r = engine(native_price=Rate()).run([paid, sold]) self.assertEqual(r["lots"][0]["source"], "derived") self.assertEqual(rep.disposal_rows(r, "FIFO")[0]["source"], "derived") class Schedule(unittest.TestCase): def test_status_is_the_first_line_of_the_summary(self): r = engine().run([buy("0xb", DAY, "1000", "10", gas="0")]) self.assertEqual(rep.summary_rows(r, {})[0]["field"], "status") def test_a_sale_the_history_cannot_cover_holds_the_report_open(self): # A scan bounded with --recent missed the purchase. The sale exported # with a zero basis, its whole proceeds as gain, and the file said COMPLETE. r = engine().run([tx("0xs", 5 * DAY, [leg(NVDA, ME, POOL, "10", 0), leg(USDG, POOL, ME, "1500", 1)], gas="0")]) self.assertEqual(r["realised"], "0.00") self.assertEqual(r["coverage"]["blocked"], 1) rows = rep.disposal_rows(r, "FIFO") self.assertEqual([(x["quantity"], x["source"], x["gain"]) for x in rows], [("10", "UNRESOLVED", rep.UNPRICED)]) self.assertTrue(summary(r)["status"].startswith("INCOMPLETE")) def test_the_schedule_adds_up_to_the_summary_with_gas(self): # Every leg carried the whole disposal's gas, and no leg subtracted it. r = engine(native_price=D("2000")).run([ buy("0xb1", 1 * DAY, "100.00", "1", gas="0.001"), buy("0xb2", 2 * DAY, "300.00", "1", gas="0.001"), tx("0xs", 9 * DAY, [leg(NVDA, ME, POOL, "2", 0), leg(USDG, POOL, ME, "800.00", 1)], gas="0.002")]) rows = rep.disposal_rows(r, "FIFO") self.assertEqual(sum(D(x["gain"]) for x in rows), D(r["realised"])) self.assertEqual(sum(D(x["gas_on_disposal"]) for x in rows), D("4.00")) self.assertEqual(D(summary(r)["realised_gain"]), D("392.00")) def test_open_lots_carry_their_exact_cost(self): # The sheet multiplied a unit cost already rounded to cents. r = engine().run([buy("0xb", DAY, "1000", "3", gas="0"), tx("0xs", 2 * DAY, [leg(NVDA, ME, POOL, "1", 0), leg(USDG, POOL, ME, "400", 1)], gas="0")]) self.assertEqual(rep.open_lot_rows(r)[0]["cost_basis"], "666.67") def test_a_wallet_move_is_recorded_on_the_lot_it_carried(self): r = engine(native_price=D("2000")).run([ buy("0xa", 100, "1000", "10", gas="0"), tx("0xmv", 200, [leg(NVDA, ME, COLD, "10")], gas="0.001"), tx("0xs", 300, [leg(NVDA, COLD, POOL, "10", 0), leg(USDG, POOL, COLD, "1500", 1)], gas="0", sender=COLD)]) self.assertEqual([(m["tx"], m["gas"]) for m in r["lots"][0]["moves"]], [("0xmv", "2.00")]) self.assertTrue(r["disposals"][0]["methods"]["FIFO"]["legs"][0]["movedWallets"]) self.assertIn("wallet move", rep.disposal_rows(r, "FIFO")[0]["note"]) def test_one_payment_for_two_assets_is_divided_not_duplicated(self): r = engine().run([tx("0xb", DAY, [leg(USDG, ME, POOL, "1000", 0), leg(NVDA, POOL, ME, "10", 1), leg(TSLA, POOL, ME, "5", 2)], gas="0")]) self.assertEqual(money(sum(D(l["cost"]) for l in r["lots"])), "1000.00") class SplitAcrossWallets(unittest.TestCase): def test_a_split_reaching_two_wallets_counts_both_and_states_what_was_open(self): r = engine().run([ buy("0xa", 1 * DAY, "5000", "50", gas="0"), tx("0xmv", 2 * DAY, [leg(NVDA, ME, COLD, "30")], gas="0"), tx("0xsplit", 3 * DAY, [leg(NVDA, ME, ZERO, "20", 0), leg(NVDA, COLD, ZERO, "30", 1), leg(NVDA, ZERO, ME, "80", 2), leg(NVDA, ZERO, COLD, "120", 3)])]) ca = r["corporateActions"][0] self.assertEqual((ca["ratio"], ca["qtyBefore"], ca["qtyAfter"], ca["costOpen"]), ("4", "50", "200", "5000.00")) self.assertIn("50 NVDA burned and 200 minted", ca["note"]) class Cli(unittest.TestCase): def test_the_rpc_flag_reaches_the_pipeline(self): seen = {} def fake_run(addresses, **kw): seen.update(kw) return {"report": None} with mock.patch.object(vrismcost, "run", fake_run): vrismcost.main(["vrismcost.py", ME, "--rpc=http://127.0.0.1:8545"]) self.assertEqual(seen.get("rpc"), "http://127.0.0.1:8545") def test_an_answer_keeps_the_date_you_give_it(self): with tempfile.TemporaryDirectory() as d: path = Path(d) / "vault.json" v, _ = Vault.create("a-long-enough-passphrase", path) v.answer("0xabc", "own", "400", date="2026-07-01") v.save() v.lock() again = Vault.unlock("a-long-enough-passphrase", path) self.assertEqual(again.answers()["0xabc"]["date"], "2026-07-01") if __name__ == "__main__": unittest.main(verbosity=2)