1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
import unittest
from decimal import Decimal
from exceptions import MismatchedTradeError, TradeNotFound
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(MismatchedTradeError):
self.lp.process_ledger([crypto_trade]) # EUR row missing
def test_deposit_notfound(self):
"""Test handling deposit with no matching withdraw."""
deposit = LedgerAction(
type="deposit",
asset="BTC",
amount=Decimal("1.00"),
fee=Decimal("0.00"),
timestamp="2025-04-17 10:00:00",
refid="67890",
)
with self.assertRaises(TradeNotFound):
self.lp.process_ledger([deposit])
def test_withdraw_afterbalance(self):
"""Test withdrawing and subsequently checking balance."""
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",
)
withdrawal = LedgerAction(
type="withdrawal",
asset="BTC",
# already includes removed fee
amount=Decimal("-0.098"),
fee=Decimal("0.001"),
timestamp="2025-04-17 12:00:00",
refid="67890",
)
self.lp.process_ledger([eur_trade, crypto_trade, withdrawal])
self.assertEqual(len(self.lp.fifo_queues["BTC"]), 0)
self.assertEqual(
self.lp.fifo_queues["BTC"].get_remaining_amount(), Decimal("0.0")
)
self.assertEqual(len(self.lp.external_wallet["BTC"]), 1)
self.assertEqual(
self.lp.external_wallet["BTC"].get_remaining_amount(), Decimal("0.098")
)
if __name__ == "__main__":
unittest.main()
|
