forked from kwikiel/rational
-
Notifications
You must be signed in to change notification settings - Fork 0
/
bank_account_test.py
147 lines (113 loc) · 4.04 KB
/
bank_account_test.py
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
139
140
141
142
143
144
145
146
147
import sys
import threading
import time
import unittest
from bank_account import BankAccount
class BankAccountTest(unittest.TestCase):
def test_newly_opened_account_has_zero_balance(self):
account = BankAccount()
account.open()
self.assertEqual(account.get_balance(), 0)
def test_can_deposit_money(self):
account = BankAccount()
account.open()
account.deposit(100)
self.assertEqual(account.get_balance(), 100)
def test_can_deposit_money_sequentially(self):
account = BankAccount()
account.open()
account.deposit(100)
account.deposit(50)
self.assertEqual(account.get_balance(), 150)
def test_can_withdraw_money(self):
account = BankAccount()
account.open()
account.deposit(100)
account.withdraw(50)
self.assertEqual(account.get_balance(), 50)
def test_can_withdraw_money_sequentially(self):
account = BankAccount()
account.open()
account.deposit(100)
account.withdraw(20)
account.withdraw(80)
self.assertEqual(account.get_balance(), 0)
def test_checking_balance_of_closed_account_throws_error(self):
account = BankAccount()
account.open()
account.close()
with self.assertRaisesWithMessage(ValueError):
account.get_balance()
def test_deposit_into_closed_account(self):
account = BankAccount()
account.open()
account.close()
with self.assertRaisesWithMessage(ValueError):
account.deposit(50)
def test_withdraw_from_closed_account(self):
account = BankAccount()
account.open()
account.close()
with self.assertRaisesWithMessage(ValueError):
account.withdraw(50)
def test_close_already_closed_account(self):
account = BankAccount()
with self.assertRaisesWithMessage(ValueError):
account.close()
def test_open_already_opened_account(self):
account = BankAccount()
account.open()
with self.assertRaisesWithMessage(ValueError):
account.open()
def test_reopened_account_does_not_retain_balance(self):
account = BankAccount()
account.open()
account.deposit(50)
account.close()
account.open()
self.assertEqual(account.get_balance(), 0)
def test_cannot_withdraw_more_than_deposited(self):
account = BankAccount()
account.open()
account.deposit(25)
with self.assertRaises(ValueError):
account.withdraw(50)
def test_cannot_withdraw_negative(self):
account = BankAccount()
account.open()
account.deposit(100)
with self.assertRaisesWithMessage(ValueError):
account.withdraw(-50)
def test_cannot_deposit_negative(self):
account = BankAccount()
account.open()
with self.assertRaisesWithMessage(ValueError):
account.deposit(-50)
def test_can_handle_concurrent_transactions(self):
account = BankAccount()
account.open()
account.deposit(1000)
self.adjust_balance_concurrently(account)
self.assertEqual(account.get_balance(), 1000)
def adjust_balance_concurrently(self, account):
def transact():
account.deposit(5)
time.sleep(0.001)
account.withdraw(5)
# Greatly improve the chance of an operation being interrupted
# by thread switch, thus testing synchronization effectively
try:
sys.setswitchinterval(1e-12)
except AttributeError:
# For Python 2 compatibility
sys.setcheckinterval(1)
threads = [threading.Thread(target=transact) for _ in range(1000)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
# Utility functions
def assertRaisesWithMessage(self, exception):
return self.assertRaisesRegex(exception, r".+")
if __name__ == '__main__':
unittest.main()