-
Notifications
You must be signed in to change notification settings - Fork 7
/
nw_test.py
92 lines (67 loc) · 3.23 KB
/
nw_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
from decimal import Decimal
import logging, sys, os
from shutil import copyfile
import sqlalchemy
from logic_bank_utils import util as logic_bank_utils
(did_fix_path, sys_env_info) = \
logic_bank_utils.add_python_path(project_dir="LogicBank", my_file=__file__)
print("\n" + did_fix_path + "\n\n" + sys_env_info + "\n\n")
from logic_bank.logic_bank import LogicBank
from logic_bank.util import row_prt, prt
from logic_bank.exec_row_logic.logic_row import LogicRow
import examples.nw.db.models as models
def copy_db_from_gold():
""" copy db/database-gold.db over db/database.db"""
print("\n" + prt("restoring database-gold\n"))
basedir = os.path.abspath(os.path.dirname(__file__))
basedir = os.path.dirname(basedir)
print("\n********************************\n"
" IMPORTANT - create database.db from database-gold.db in " + basedir + "/payment_allocation/db/\n" +
" - from -- " + prt("") +
"\n********************************")
db_loc = os.path.join(basedir, "db/database.db")
db_source = os.path.join(basedir, "db/database-gold.db")
copyfile(src=db_source, dst=db_loc)
def setup_logging():
logic_logger = logging.getLogger('logic_logger') # for debugging user logic
logic_logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(message)s - %(asctime)s - %(name)s - %(levelname)s')
handler.setFormatter(formatter)
logic_logger.addHandler(handler)
do_engine_logging = False
engine_logger = logging.getLogger('engine_logger') # for internals
if do_engine_logging:
engine_logger.setLevel(logging.DEBUG)
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(message)s - %(asctime)s - %(name)s - %(levelname)s')
handler.setFormatter(formatter)
engine_logger.addHandler(handler)
setup_logging()
copy_db_from_gold()
basedir = os.path.abspath(os.path.dirname(__file__))
basedir = os.path.dirname(basedir)
db_loc = os.path.join(basedir, "db/database.db")
conn_string = "sqlite:///" + db_loc
engine = sqlalchemy.create_engine(conn_string, echo=False) # sqlalchemy sqls...
session_maker = sqlalchemy.orm.sessionmaker()
session_maker.configure(bind=engine)
session = session_maker()
from examples.nw.logic.logic import declare_logic
LogicBank.activate(session=session, activator=declare_logic)
"""
Not regression test - isolated test for logic bank debugging
"""
new_employee = models.Employee(LastName='Obama', Salary=100000, WorksFor=1, OnLoan=2, IsCommissioned=0) # project_id (fk, not id) triggers clone
session.add(new_employee)
session.commit()
print("\nadd_employee, update completed\n\n")
works_for = new_employee.Works_for_dept
assert works_for.Name == "Sales", f'Expected Sales, got {works_for.Name}'
assert works_for.SalaryTotal == 283000, f'Expected SalaryTotal 283000, got {works_for.SalaryTotal}'
assert works_for.WorksForCount == 3, f'Expected WorksForCount == 3, got {works_for.WorksForCount}'
on_loan = new_employee.On_loan_dept
assert on_loan.OnLoanCount == 1, f'Expected OnLoanCount == 1, got {on_loan.OnLoanCount}'
print("\n...new_employee, ran to completion\n\n")