diff options
| -rw-r--r-- | ledger_process.py | 16 | ||||
| -rw-r--r-- | test_ledger_process.py | 68 |
2 files changed, 70 insertions, 14 deletions
diff --git a/ledger_process.py b/ledger_process.py index 24a6aca..dd82877 100644 --- a/ledger_process.py +++ b/ledger_process.py @@ -37,25 +37,11 @@ class LedgerProcess: # Group trades by refid if action.type == "trade": - assert len(actions) == 2 self._process_trade(refid, actions) elif action.type == "deposit" and action.asset != "EUR": assert len(actions) == 1 logger.error("Don't know how do handle deposits yet.") - # currency = action.asset - # fifo_queues.setdefault(currency, FIFOQueue()) - # amount = Decimal(action.amount) - # price = 0 - # current = fifo_queues[currency] - - # # remove transaction fees - # # (but only if it has a previous withdraw, or rather, if these are coins from a previous purchase... - # # but I can't check that...) - # if len(current): - # current.remove_coins(action.fee) - - # current.add(amount, price, action.date) elif action.type == "withdrawal" and action.asset != "EUR": assert len(actions) == 1 @@ -111,6 +97,8 @@ class LedgerProcess: Trade(crypto_amount, -proceeds, date_sold, refid=refid) ) else: + logger.error(f"Trade group doesn't have expected currencies.") raise ValueError(f"Unexpected trade grouping for refid {refid}") else: + logger.error(f"Trade group has {len(trades)} trades, expected 2.") raise ValueError(f"Unexpected number of trades for refid {refid}") diff --git a/test_ledger_process.py b/test_ledger_process.py new file mode 100644 index 0000000..092619a --- /dev/null +++ b/test_ledger_process.py @@ -0,0 +1,68 @@ +import unittest +from decimal import Decimal +from ledger_action import LedgerAction +from ledger_process import LedgerProcess + + +class TestLedgerProcess(unittest.TestCase): + def setUp(self): + """Set up a LedgerProcess instance for testing.""" + self.lp = LedgerProcess() + + def test_process_trade_valid(self): + """Test valid processing of a trade with both EUR and crypto assets.""" + eur_trade = LedgerAction( + type="trade", + asset="EUR", + amount=Decimal("-500.00"), + fee=Decimal("2.00"), + timestamp="2025-04-17 10:00:00", + refid="12345", + ) + crypto_trade = LedgerAction( + type="trade", + asset="BTC", + amount=Decimal("0.1"), + fee=Decimal("0.001"), + timestamp="2025-04-17 10:00:00", + refid="12345", + ) + self.lp.process_ledger([eur_trade, crypto_trade]) + # Assert the remaining balance in the FIFO queue + self.assertEqual( + self.lp.fifo_queues["BTC"].get_remaining_amount(), Decimal("0.099") + ) + + def test_empty_actions(self): + """Test processing with no actions.""" + self.lp.process_ledger([]) + # with self.assertRaises(ValueError): + + def test_missing_trade_rows(self): + """Test processing a trade with missing EUR or crypto rows.""" + crypto_trade = LedgerAction( + type="trade", + asset="BTC", + amount=Decimal("0.1"), + fee=Decimal("0.001"), + timestamp="2025-04-17 10:00:00", + refid="12345", + ) + with self.assertRaises(ValueError): + self.lp.process_ledger([crypto_trade]) # EUR row missing + + def test_unsupported_deposit(self): + """Test handling unsupported deposits.""" + deposit = LedgerAction( + type="deposit", + asset="BTC", + amount=Decimal("1.00"), + fee=Decimal("0.00"), + timestamp="2025-04-17 10:00:00", + refid="67890", + ) + self.lp.process_ledger([deposit]) + + +if __name__ == "__main__": + unittest.main() |