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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
import os
import tempfile
import unittest
from decimal import Decimal
from unittest.mock import mock_open, patch
from kraken import parse_row, read_ledger
from ledger_action import LedgerAction
class TestKrakenFunctions(unittest.TestCase):
def test_parse_kraken_row_valid_input(self):
row = {
"type": "deposit",
"asset": "BTC",
"amount": "0.5",
"fee": "0.001",
"time": "2025-04-16 12:00:00",
"refid": "12345",
}
expected = LedgerAction(
type="deposit",
asset="BTC",
amount=Decimal("0.5"),
fee=Decimal("0.001"),
refid="12345",
timestamp="2025-04-16 12:00:00",
)
self.assertEqual(parse_row(row), expected)
def test_parse_kraken_row_missing_fields(self):
row = {
"type": "trade",
"asset": "ETH",
"amount": "2.0",
"time": "2025-04-16 15:00:00",
}
expected = LedgerAction(
type="trade",
asset="ETH",
amount=Decimal("2.0"),
fee=Decimal("0"), # Default fee
refid="", # Default refid
timestamp="2025-04-16 15:00:00",
)
self.assertEqual(parse_row(row), expected)
@patch(
"builtins.open",
new_callable=mock_open,
read_data='"txid","refid","time","type","subtype","aclass","asset","wallet","amount","fee","balance"\n"bla","67890","2025-04-16 09:00:00","trade","bla","currency","BTC","main","1.5","0.01"\n',
)
def test_read_kraken_ledger(self, mock_file):
expected = [
LedgerAction(
type="trade",
asset="BTC",
amount=Decimal("1.5"),
fee=Decimal("0.01"),
refid="67890",
timestamp="2025-04-16 09:00:00",
)
]
self.assertEqual(read_ledger("dummy_path.csv"), expected)
class TestKrakenFunctionsRealFiles(unittest.TestCase):
def setUp(self):
# Create a temporary CSV file for tests
self.temp_file = tempfile.NamedTemporaryFile(
delete=False, mode="w", suffix=".csv"
)
self.temp_file.write(
'"txid","refid","time","type","subtype","aclass","asset","wallet","amount","fee","balance"\n'
'"","","2024-07-01 00:00:00","deposit","","currency","EUR","spot / main",1000.0000,0,1000.0000\n'
'"","d1e57f","2024-07-12 16:26:22","trade","tradespot","currency","EUR","spot / main",-130.4204,0,0.0000\n'
'"","d1e57f","2024-07-12 16:26:22","trade","tradespot","currency","DOGE","spot / main",1321.95097670,5.28780391,1316.66317279\n'
'"","","2024-07-12 16:36:49","withdrawal","","currency","DOGE","spot / main",-1312.66317279,4.00000000,0.00000000\n'
'"","","2024-08-02 14:24:30","deposit","","currency","EUR","spot / main",100.0000,0,100.0000\n'
)
self.temp_file.close() # Close the file so it can be read later
def tearDown(self):
# Remove the temporary file after tests are done
os.unlink(self.temp_file.name)
def test_read_kraken_ledger_with_real_file(self):
# Define the expected result
expected = [
LedgerAction(
type="deposit",
asset="EUR",
amount=Decimal("1000.0000"),
fee=Decimal("0"),
refid="",
timestamp="2024-07-01 00:00:00",
),
LedgerAction(
type="trade",
asset="EUR",
amount=Decimal("-130.4204"),
fee=Decimal("0"),
refid="d1e57f",
timestamp="2024-07-12 16:26:22",
),
LedgerAction(
type="trade",
asset="DOGE",
amount=Decimal("1321.95097670"),
fee=Decimal("5.28780391"),
refid="d1e57f",
timestamp="2024-07-12 16:26:22",
),
LedgerAction(
type="withdrawal",
asset="DOGE",
amount=Decimal("-1312.66317279"),
fee=Decimal("4.00000000"),
refid="",
timestamp="2024-07-12 16:36:49",
),
LedgerAction(
type="deposit",
asset="EUR",
amount=Decimal("100.0000"),
fee=Decimal("0"),
refid="",
timestamp="2024-08-02 14:24:30",
),
]
# Test the function with the temporary file
self.assertEqual(read_ledger(self.temp_file.name), expected)
if __name__ == "__main__":
unittest.main()
|
