-
Notifications
You must be signed in to change notification settings - Fork 8
/
Copy pathtest_add_order.py
208 lines (174 loc) · 9.22 KB
/
test_add_order.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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
import sys, unittest
import logic_bank_utils.util as logic_bank_utils
from datetime import datetime
(did_fix_path, sys_env_info) = \
logic_bank_utils.add_python_path(project_dir="LogicBank", my_file=__file__)
if __name__ == '__main__':
print("\nStarted from cmd line - launch unittest and exit\n")
sys.argv = [sys.argv[0]]
unittest.main(module="examples.nw.tests.test_add_order")
exit(0)
else:
print("Started from unittest: " + __name__)
from examples.nw import tests
tests.copy_gold_over_db()
import examples.nw.db.models as models
from examples.nw.logic import session, engine # opens db, activates rules <--
# activate rules: LogicBank.activate(session=session, activator=declare_logic)
from logic_bank.exec_row_logic.logic_row import LogicRow # must follow import of models
from logic_bank.util import prt, row_prt, ConstraintException
print("\n" + sys_env_info + "\n\n")
class Test(unittest.TestCase):
def setUp(self): # banner
self.started_at = str(datetime.now())
tests.setUp(file=__file__)
def tearDown(self):
tests.tearDown(file=__file__, started_at=self.started_at, engine=engine, session=session)
def test_run(self):
pre_cust = session.query(models.Customer).filter(models.Customer.Id == "ALFKI").one()
session.expunge(pre_cust)
# debug switches to bypass tests (normally all true)
test1 = False # Fixme re-enable test
test2 = False
test3 = False
test4 = True
"""
Test 1 - should fail due to credit limit exceeded
"""
if test1:
bad_order = models.Order(AmountTotal=0, CustomerId="ALFKI", ShipCity="Richmond",
EmployeeId=6, Freight=1)
session.add(bad_order)
# OrderDetails - https://docs.sqlalchemy.org/en/13/orm/backref.html
bad_item1 = models.OrderDetail(ProductId=1, Amount=0,
Quantity=1, UnitPrice=18,
Discount=0)
bad_order.OrderDetailList.append(bad_item1)
bad_item2 = models.OrderDetail(ProductId=2, Amount=0,
Quantity=20000, UnitPrice=18,
Discount=0)
bad_order.OrderDetailList.append(bad_item2)
did_fail_as_expected = False
try:
session.commit()
except ConstraintException as ce:
print("Expected constraint: " + str(ce))
session.rollback()
did_fail_as_expected = True
except:
self.fail("Unexpected Exception Type")
if not did_fail_as_expected:
self.fail("huge order expected to fail, but succeeded")
else:
print("\n" + prt("huge order failed credit check as expected. Now trying non-commissioned order, should also fail..."))
"""
Test 2 - should fail due to not-commissioned
"""
if test2:
bad_order = models.Order(AmountTotal=0, CustomerId="ALFKI", ShipCity="Richmond",
EmployeeId=2, Freight=1)
session.add(bad_order)
# OrderDetails - https://docs.sqlalchemy.org/en/13/orm/backref.html
bad_item1 = models.OrderDetail(ProductId=1, Amount=0,
Quantity=1, UnitPrice=18,
Discount=0)
bad_order.OrderDetailList.append(bad_item1)
bad_item2 = models.OrderDetail(ProductId=2, Amount=0,
Quantity=2, UnitPrice=18,
Discount=0)
bad_order.OrderDetailList.append(bad_item2)
did_fail_as_expected = False
try:
session.commit()
except ConstraintException:
session.rollback()
did_fail_as_expected = True
except:
print("Unexpected Exception Type")
if not did_fail_as_expected:
self.fail("order for non-commissioned expected to fail, but succeeded")
else:
print("\n" + prt("non-commissioned order failed constraint as expected. Now trying invalid customer, should fail..."))
"""
Test 3 - should fail due to invalid customer
"""
if test3:
bad_order = models.Order(AmountTotal=0, CustomerId="XX INVALID CUSTOMER", ShipCity="Richmond",
EmployeeId=6, Freight=1)
session.add(bad_order)
# OrderDetails - https://docs.sqlalchemy.org/en/13/orm/backref.html
bad_item1 = models.OrderDetail(ProductId=1, Amount=0,
Quantity=1, UnitPrice=18,
Discount=0)
bad_order.OrderDetailList.append(bad_item1)
bad_item2 = models.OrderDetail(ProductId=2, Amount=0,
Quantity=2, UnitPrice=18,
Discount=0)
bad_order.OrderDetailList.append(bad_item2)
did_fail_as_expected = False
try:
session.commit()
except ConstraintException as ce:
session.rollback()
did_fail_as_expected = True
print(str(ce))
except:
session.rollback()
did_fail_as_expected = True
e = sys.exc_info()[0]
print(e)
if not did_fail_as_expected:
self.fail("order for invalid customer expected to fail, but succeeded")
else:
print("\n" + prt("invalid customer failed as expected. Now trying valid order, should succeed..."))
"""
Test 4 - should succeed
"""
if test4: # NB: use non-0 AmountTotal to verify aggregate_defaults
# dragon alert - this can process OrderHeader first in log, or OrderDetails.
# run by hand a few times to ensure both run
new_order = models.Order(AmountTotal=101, CustomerId="ALFKI", ShipCity="Richmond",
EmployeeId=6, Freight=1)
new_order = models.Order(AmountTotal=0, CustomerId="ALFKI", ShipCity="Richmond",
EmployeeId=6, Freight=1)
session.add(new_order)
# OrderDetails - https://docs.sqlalchemy.org/en/13/orm/backref.html
new_item1 = models.OrderDetail(ProductId=1, Amount=0,
Quantity=1, UnitPrice=18,
Discount=0)
new_order.OrderDetailList.append(new_item1)
new_item2 = models.OrderDetail(ProductId=2, Amount=0,
Quantity=2, UnitPrice=18,
Discount=0)
new_order.OrderDetailList.append(new_item2)
session.commit()
post_cust = session.query(models.Customer).filter(models.Customer.Id == "ALFKI").one()
print("\nadd_order, update completed - analyzing results..\n\n")
row_prt(new_order, session, "\nnew Order Result (seeking $56)") # $18 + $38 = $56
if new_order.AmountTotal != 56:
self.fail(row_prt(new_order, "Unexpected AmountTotal (seeking $56): " + str(new_order.AmountTotal) +
"... expected 56"))
row_prt(new_item1, session, "\nnew Order Detail 1 Result") # 1 Chai @ $18
row_prt(new_item2, session, "\nnew Order Detail 2 Result") # 2 Chang @ $19 = $38
logic_row = LogicRow(row=post_cust, old_row=pre_cust, ins_upd_dlt="*", nest_level=0, a_session=session, row_sets=None)
if post_cust.Balance == pre_cust.Balance + 56:
logic_row.log("Correct adjusted Customer Result")
assert True
else:
self.fail(logic_row.log(f"ERROR - incorrect adjusted Customer Result {post_cust.Balance} != {pre_cust.Balance + 56}"))
if post_cust.OrderCount == pre_cust.OrderCount + 1 and\
post_cust.UnpaidOrderCount == pre_cust.UnpaidOrderCount + 1:
pass
else:
self.fail(logic_row.log("Error - unexpected OrderCounts - did not increase by 1"))
from sqlalchemy.sql import func
qry = session.query(models.Order.CustomerId,
func.sum(models.Order.AmountTotal).label('sql_balance'))\
.filter(models.Order.CustomerId == "ALFKI", (models.Order.ShippedDate == '') | (models.Order.ShippedDate == None))
qry = qry.group_by(models.Order.CustomerId).one()
if qry.sql_balance == post_cust.Balance:
logic_row.log(f"Derived balance matches sql `select sum' result {qry.sql_balance}: " + str(post_cust.Balance))
else:
self.fail(logic_row.log(f"ERROR - computed balance {str(post_cust.Balance)} does not match sql result: {qry.sql_balance}"))
print("\nadd_order, ran to completion\n\n")
self.assertTrue(True)