solar - no engine. Simplified test setUp/tearDown

logic_bank/exec_row_logic/logic_row.py

@@ -54,8 +54,8 @@ def __init__(self, row: base, old_row: base, ins_upd_dlt: str, nest_level: int,
 
         rb = RuleBank()
         self.rb = rb
-        self.session = rb._session
-        # solar no engine self.engine = rb._engine
+        self.session = a_session
+        # solar no engine self.session = rb.session; self.engine = rb._engine
         self.some_base = declarative_base()
 
         self.name = type(self.row).__name__  # class name (not table name)

logic_bank/logic_bank.py

@@ -4,7 +4,7 @@
 from sqlalchemy.orm import session
 
 from logic_bank.rule_bank import rule_bank_withdraw  # reduce circular imports
-from logic_bank.rule_bank.rule_bank_setup import setup, compute_formula_execution_order
+import logic_bank.rule_bank.rule_bank_setup as rule_bank_setup
 from logic_bank.rule_type.constraint import Constraint
 from logic_bank.rule_type.copy import Copy
 from logic_bank.rule_type.count import Count
@@ -17,24 +17,27 @@ class LogicBank:
     """
     Logic consists of Rules, and Python.
 
-    Activate your logic,
-    providing a function that declares your rules and Python.
+    Activate your logic by calling
+
+        activate(session: session, activator: my_logic)
+
+    where myLogic is a function that declares your rules and Python.
     """
 
     def activate(session: session, activator: callable):
         """
-        load rules - executed on commit
+        load rules - later executed on commit
 
         raises exception if cycles detected
 
         :param session: SQLAlchemy session
         :param activator: function that declares rules (e.g., Rule.sum...)
-        :return:
         """
+
         # solar engine = session.bind.engine
-        setup(session)  # solar , engine)
+        rule_bank_setup.setup(session)  # solar , engine)
         activator()
-        compute_formula_execution_order()   # solar session, engine)
+        rule_bank_setup.compute_formula_execution_order()   # solar session, engine)
 
 
 class Rule:

logic_bank/rule_bank/rule_bank_setup.py

@@ -11,11 +11,13 @@
 
 def setup(a_session: session):  # solar , an_engine: Engine):
     """
-    Initialize the RuleBank
+    Create the RuleBank
+
+    Register before_flush listeners
 
     """
     rules_bank = RuleBank()
-    rules_bank._session = a_session
+    # rules_bank._session = a_session  """ solar no rb.session... """
     event.listen(a_session, "before_flush", before_flush)
     event.listen(a_session, "before_commit", before_commit)
 

nw/basic_web_app/app/__init__.py

@@ -1,6 +1,4 @@
 import logging
-import os
-import sys
 
 from flask import Flask
 from flask_appbuilder import AppBuilder, SQLA
@@ -10,35 +8,15 @@
 use_rules = True
 
 if use_rules:
-    cwd = os.getcwd()  # eg, /Users/val/python/pycharm/logic-bank/basic_web_app
-    required_path_python_rules = cwd  # seeking /Users/val/python/pycharm/logic-bank
-    required_path_python_rules = required_path_python_rules.replace("/nw/basic_web_app", "")
-    required_path_python_rules = required_path_python_rules.replace("\\nw\\basic_web_app", "")
-    required_path_python_rules = required_path_python_rules.replace("\\\\", "\\")  # you cannot be serious
-
-    sys_path = ""
-    required_path_present = False
-    for each_node in sys.path:
-        sys_path += each_node + "\n"
-        if each_node == required_path_python_rules:
-            required_path_present = True
-    print("\n sys.path...\n" + sys_path)
-    if not required_path_present:
-        print("basic_web_app/app/__init__.py fixing path (so can run from terminal) with: " +
-              required_path_python_rules)
-        sys.path.append(required_path_python_rules)
-        print("sys_path: " + str(sys.path))
-    else:
-        pass
-        print("NOT Fixing path (default PyCharm, set in VSC Launch Config): " +
-              required_path_python_rules)
+    import logic_bank_utils.util as logic_bank_utils
+    (did_fix_path, sys_env_info) = \
+        logic_bank_utils.add_python_path(project_dir="LogicBank", my_file=__file__)
 
     import nw.db.models as models  # FIXME design prevents circular imports
 
-    from logic_bank.rule_bank import rule_bank_setup
-    from nw.logic import declare_logic
+    from nw.logic.rules_bank import declare_logic
 
-    from logic_bank.logic_bank import LogicBank
+    from logic_bank.logic_bank import LogicBank  # activate rules (calls declare_logic)
 
 """
  Logging configuration

nw/db/__init__.py

@@ -0,0 +1,24 @@
+import os
+import sqlalchemy
+from sqlalchemy.orm import session
+
+
+class DB:
+    """
+    Opens the database
+
+    Sets instance variables session, engine
+    """
+    def __init__(self):
+        # def open_database() -> (sqlalchemy.orm.session.Session, sqlalchemy.engine.base.Engine):
+        basedir = os.path.abspath(os.path.dirname(__file__))
+        basedir = os.path.dirname(basedir)
+
+        nw_loc = os.path.join(basedir, "db/database.db")
+
+        conn_string = "sqlite:///" + nw_loc
+        self.engine = sqlalchemy.create_engine(conn_string, echo=False)  # sqlalchemy sqls...
+
+        session_maker = sqlalchemy.orm.sessionmaker()
+        session_maker.configure(bind=self.engine)
+        self.session = session_maker()

nw/logic/__init__.py

@@ -1,68 +0,0 @@
-import os
-
-import sqlalchemy
-from sqlalchemy.orm import session
-
-from logic_bank.logic_bank import LogicBank
-from nw.logic.rules_bank import declare_logic
-
-import nw.logic.legacy.setup as legacy_setup
-
-from logic_bank.util import prt
-
-""" Initialization
-1 - Connect
-2 - Register listeners (either hand-coded ones above, or the logic-engine listeners).
-"""
-
-print("\n")
-print("*********************")
-print(prt("BEGIN - setup logging, connect to db, register listeners"))
-print("*********************")
-
-# Initialize Logging
-import logging
-import sys
-
-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)
-
-basedir = os.path.abspath(os.path.dirname(__file__))
-basedir = os.path.dirname(basedir)
-
-nw_loc = os.path.join(basedir, "db/database.db")
-
-conn_string = "sqlite:///" + nw_loc
-engine = sqlalchemy.create_engine(conn_string, echo=False)  # sqlalchemy sqls...
-
-session_maker = sqlalchemy.orm.sessionmaker()
-session_maker.configure(bind=engine)
-session = session_maker()
-
-by_rules = True  # True => use rules, False => use legacy hand code (for comparison)
-rule_list = None
-db = None
-if by_rules:
-    LogicBank.activate(session=session, activator=declare_logic)
-else:
-    legacy_setup.setup(session)  # ignore test asserts that fail due to (unimplemenmted) counts (else ok)
-
-print("\n")
-print("*********************")
-print(prt("END - connected, session created, listeners registered"))
-print("*********************")

nw/tests/__init__.py

@@ -10,6 +10,9 @@
 def copy_gold_over_db():
     """ copy db/database-gold.db over db/database.db"""
 
+    # import time
+    # time.sleep(1)
+
     basedir = os.path.abspath(os.path.dirname(__file__))
     basedir = os.path.dirname(basedir)
 
@@ -22,32 +25,85 @@ def copy_gold_over_db():
     nw_source = os.path.join(basedir, "db/database-gold.db")
     copyfile(src=nw_source, dst=nw_loc)
 
+def setup_logging():
+    # Initialize Logging
+    import logging
+    import sys
+
+    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)
+
+
+def setUp(test: object, file: str):
+    """
+    SETUP - logging, connect to db, register listeners, activate logic
+    """
+
+    print("\n")
+    print("*********************")
+    print("BEGIN SETUP - logging, connect to db, register listeners, activate logic")
+    print("*********************")
 
-def setUp(file: str):
     copy_gold_over_db()
 
+    setup_logging()
+
+    from logic_bank.logic_bank import LogicBank
+    from nw.logic.rules_bank import declare_logic
+
+    import nw.logic.legacy.setup as legacy_setup
+
+    import nw.db as open_db
+    test.db = open_db.DB()
+    test.session = test.db.session
+    test.engine = test.db.engine
+
+    by_rules = True  # True => use rules, False => use legacy hand code (for comparison)
+    if by_rules:
+        LogicBank.activate(session=test.session, activator=declare_logic)
+    else:
+        legacy_setup.setup(test.session)  # ignore test asserts that fail due to (unimplemented) counts (else ok)
+
     print("\n")
     print("**********************")
-    print("** Setup complete - test execution begins for: " + file)
+    print("** END SETUP - logging, database and logic are setup")
+    print("** Test execution begins for: " + file)
+    print("** Session: " + str(test.session))
     print("** Started: " + str(datetime.now()))
     print("** Following log best viewed without word wrap")
     print("**********************")
+    print("\n")
 
 
-def tearDown(file: str, started_at: str, engine: sqlalchemy.engine.base.Engine, session: sqlalchemy.orm.session.Session):
+def tearDown(file: str, started_at: str, test: object):
     """
     close session & engine, banner
 
     :param file: caller, usually __file__
     :param started_at: eg, str(datetime.now())
-    :param engine: eg, nw.logic import session, engine
-    :param session: from nw.logic import session, engine
+    :param test: test instance
     :return:
     """
-    session.close()
-    engine.dispose()
+    test.session.close()
+    test.engine.dispose()
     print("\n")
     print("**********************")
-    print("** Test complete, SQLAlchemy session/engine closed for: " + file)
+    print("** Test tearDown complete, SQLAlchemy session/engine closed for: " + file)
+    print("** Session: " + str(test.session))
     print("** Started: " + started_at + " Ended: " + str(datetime.now()))
     print("**********************")

nw/tests/test_add_cust.py

@@ -14,42 +14,44 @@
     print("Started from unittest: " + __name__)
     import nw.tests as tests  # careful - this must follow add_python_path, above
 
-    tests.copy_gold_over_db()
-
     import nw.db.models as models
-    from nw.logic import session, engine  # opens db, activates rules <--
 
     from logic_bank.exec_row_logic.logic_row import LogicRow  # must follow import of models
-    from logic_bank.util import prt, row_prt
+    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__)
+        self.session = None
+        self.engine = None
+
+        tests.setUp(test=self, file=__file__)
+        pass
 
     def tearDown(self):
-        tests.tearDown(file=__file__, started_at=self.started_at, engine=engine, session=session)
+        tests.tearDown(file=__file__, started_at=self.started_at, test=self)
 
     def test_run(self):
         # first delete, so can add
-        delete_cust = session.query(models.Customer).filter(models.Customer.Id == "$$New Cust").delete()
+        delete_cust = self.session.query(models.Customer).filter(models.Customer.Id == "$$New Cust").delete()
         print("\nadd_cust, deleting: " + str(delete_cust) + "\n\n")
-        session.commit()
+        self.session.commit()
 
         # Add a Customer - works
         new_cust = models.Customer(Id="$$New Cust", Balance=0, CreditLimit=0)
-        session.add(new_cust)
-        session.commit()
+        self.session.add(new_cust)
+        self.session.commit()
 
-        verify_cust = session.query(models.Customer).filter(models.Customer.Id == "$$New Cust").one()
+        verify_cust = self.session.query(models.Customer).filter(models.Customer.Id == "$$New Cust").one()
 
         print("\nadd_cust, verified: " + str(verify_cust) + "\n\n")
 
         from sqlalchemy.sql import func
-        qry = session.query(models.Order.CustomerId, func.sum(models.Order.AmountTotal))\
+        qry = self.session.query(models.Order.CustomerId, func.sum(models.Order.AmountTotal))\
             .filter(models.Order.CustomerId == "ALFKI", models.Order.ShippedDate == None)
         qry = qry.group_by(models.Order.CustomerId)
         for _res in qry.all():