details for code elegance

Backend/Sources/Modules/Inventory/inventory.py

@@ -129,7 +129,6 @@ def get_all_inventory_items() :
 
     with persistence.get_Session_Instance() as sess :
         Items = sess.query(inventory_item).order_by(inventory_item.id)
-
         return Items
 
 
@@ -201,12 +200,13 @@ def inventory_group_add_item(group_id,inventory_id) :
 def inventory_group_delete_item(group_id,inventory_id) :
     with persistence.get_Session_Instance() as sess :
 
-        group = sess.query(inventory_group).filter(inventory_group.id == group_id).first()
-        Item = sess.query(inventory_item).filter(inventory_item.id == inventory_id).first()
+        group : inventory_group = sess.query(inventory_group).filter(inventory_group.id == group_id).first()
+        Item  : inventory_item = sess.query(inventory_item).filter(inventory_item.id == inventory_id).first()
         if group_id == Item.group_id :
             Item.group_id = None
 
-        #TODO delete group if orphan ? maybe configure in model
+        if len(sess.query(inventory_item).filter(inventory_item.group_id == group.id).first()) == 0 :
+            sess.delete(group) # if group is orphan
         sess.commit()
 
 def inventory_group_delete(group_id) :

Backend/Sources/Modules/Inventory/inventory_view.py

@@ -10,12 +10,12 @@ import Modules.Inventory.inventory as inventory
 
 
 from flask import Blueprint
-inventory_blueprint = Blueprint('Inventory', __name__,url_prefix='/inventory')
+inventory_blueprint = Blueprint('Inventory', __name__,url_prefix='/api/inventory')
 
 
 
-inventory_read_only_role  = Privilege_Role(name="read_only")
-inventory_admin_role      = Privilege_Role(name="admin")
+inventory_read_only_role  = Privilege_Role(name="Read Only")
+inventory_admin_role      = Privilege_Role(name="Admin")
 
 
 def init():
@@ -97,6 +97,13 @@ def declare_route() :
         return model_manager.ModelObjectToJsonString(inventory.inventory_item_create(json.dumps(flask.request.json)))
 
 
+    @inventory_blueprint.route('/item/template',methods = ['GET'])
+    @view_privilege.manager.require_authorization(required_role=inventory_read_only_role,get_privilege_func=view.get_user_privilege)
+    def get_item_template() :
+
+        return model_manager.ModelObjectToJsonString(inventory.inventory_item())
+
+
 
 #############################################
 ########           sites             ########
@@ -131,6 +138,12 @@ def declare_route() :
     def create_site() :
         return model_manager.ModelObjectToJsonString(inventory.inventory_site_create(json.dumps(flask.request.json)))
 
+    @inventory_blueprint.route('/site/template',methods = ['GET'])
+    @view_privilege.manager.require_authorization(required_role=inventory_admin_role,get_privilege_func=view.get_user_privilege)
+    def get_site_template() :
+
+        return model_manager.ModelObjectToJsonString(inventory.inventory_site())
+
 #############################################
 ########           contacts          ########
 #############################################
@@ -165,6 +178,13 @@ def declare_route() :
     def create_contact() :
         return model_manager.ModelObjectToJsonString(inventory.inventory_contact_create(json.dumps(flask.request.json)))
 
+
+    @inventory_blueprint.route('/contact/template',methods = ['GET'])
+    @view_privilege.manager.require_authorization(required_role=inventory_admin_role,get_privilege_func=view.get_user_privilege)
+    def get_contact_template() :
+
+        return model_manager.ModelObjectToJsonString(inventory.inventory_contact())
+
 #############################################
 ########           groups            ########
 #############################################
@@ -207,6 +227,13 @@ def declare_route() :
             return model_manager.ModelObjectToJsonString(inventory.inventory_group_delete_item(groupID,itemID))
 
 
+    @inventory_blueprint.route('/group/template',methods = ['GET'])
+    @view_privilege.manager.require_authorization(required_role=inventory_admin_role,get_privilege_func=view.get_user_privilege)
+    def get_group_template() :
+
+        return model_manager.ModelObjectToJsonString(inventory.inventory_group())
+
+
 
 
 

Backend/Sources/View/view_error_management.py

@@ -2,6 +2,7 @@ import View.view_privilege as privileges
 import logging
 from flask import jsonify
 from utility.app_logging import logger_name
+import sys
 logger = logging.getLogger(logger_name + ".VIEW")
 
 import traceback
@@ -23,7 +24,8 @@ def define_error_management(app) :
 
 
     @app.errorhandler(Exception)
-    def handle_privilege_error(err) :
+    def handle_generic_error(err) :
+        exc_type,exc_obj,exc_tb = sys.exc_info()
         """Return JSON instead of HTML for any other server error"""
         logger.error(f"Exception: {str(err)}")
         response = {"error": str(err) }

Backend/Sources/View/view_manager.py

@@ -7,14 +7,14 @@ import logging
 import persistence
 import Model.isp_model as isp_model
 import Model.model_manager as model_manager
-from werkzeug.security import check_password_hash
+from werkzeug.security import check_password_hash,generate_password_hash
 import View.view_privilege as privileges
 import Modules.Inventory.inventory_view as inventory_view
 from datetime import timedelta
 import View.view_error_management as view_error_management
 from flask_limiter import Limiter
 from flask_limiter.util import get_remote_address
-from flask_cors import CORS
+
 
 logger = logging.getLogger(logger_name + ".VIEW")
 
@@ -32,6 +32,7 @@ limiter = Limiter(__app__,key_func=get_remote_address,default_limits=["500 per m
 limiter.logger = logger
 
 
+
 from werkzeug.serving import make_server
 class ServerThread(threading.Thread):
 
@@ -95,7 +96,7 @@ def after_request(response):
     # for exemple origin cross origin is when website with javascript has it's server (origin 1)
     # and the javascript call some request on another server (origin 2), typically our API.
     header['Access-Control-Allow-Origin'] = '*'
-    header['Access-Control-Allow-Methods'] = 'GET,HEAD,OPTIONS,POST,PUT'
+    header['Access-Control-Allow-Methods'] = 'GET,DELETE,UPDATE,HEAD,OPTIONS,POST,PUT'
     header['Access-Control-Allow-Headers'] = 'Origin, X-Requested-With, Content-Type, Accept, Authorization'
     return response
 
@@ -132,7 +133,6 @@ def login():
         resp.status_code = 200
         return resp
 
-
 @__app__.route('/api/logout',methods = ['DELETE'])
 def logout():
     logger.info("account " + session["username"] + " logged OUT with id : " + session["client_id"])
@@ -140,6 +140,24 @@ def logout():
     return jsonify('logout')
 
 
+@__app__.route('/api/password',methods = ['POST'])
+def change_password():
+    _json = request.json
+    _old_password = _json['old_password']
+    _password     = _json['new_password']
+    with persistence.get_Session_Instance() as sess :
+        Item : isp_model.user_account = sess.query(isp_model.user_account).filter(isp_model.user_account.id == session["user_account_id"]).first()
+
+
+        if not check_password_hash(Item.password,_password) :
+            raise Exception("old password is incorrect")
+
+        Item.password = generate_password_hash(_password)
+        sess.commit()
+
+    return jsonify('password changed')
+
+
 @__app__.route('/routes',methods = ['GET'])
 @privileges.manager.require_authorization(required_role=inventory_view.inventory_admin_role,get_privilege_func=get_user_privilege)
 def routes():

Backend/Sources/config_manager.py

@@ -0,0 +1,29 @@
+import configparser
+from pathlib import Path
+path_to_file = "configuration.ini"
+
+
+def create_conf_file() :
+    parser = configparser.ConfigParser()
+    parser.add_section('core')
+    parser.set('core', 'default_admin_password', 'aseqzdwxc')
+    parser.add_section('log')
+    parser.set('log', 'console_log_level', 'DEBUG')
+    parser.set('log', 'file_log_level', 'DEBUG')
+
+    fp=open(path_to_file,'w')
+    parser.write(fp)
+    fp.close()
+
+
+
+path = Path(path_to_file)
+
+if path.is_file() :
+    None
+else :
+    create_conf_file()
+
+
+
+

Backend/Sources/main.py

@@ -1,4 +1,5 @@
 
+from concurrent.futures import thread
 from pyexpat import model
 from time import sleep
 
@@ -8,11 +9,15 @@ import persistence
 from utility.app_logging import logger_name, init as init_logging
 import logging
 import View.view_manager as view
-
+import threading
 
 logger = logging.getLogger(logger_name)
 __is_running__ = False
 __stop_app__ = False
+condition_object = threading.Condition()
+
+
+
 def init() :
 
 
@@ -69,26 +74,31 @@ def main() :
     global __is_running__
     global __stop_app__
     __is_running__ = True
-
-
+    condition_object.acquire()
     logger.info("Open ISP is now Running ! ")
     while(not __stop_app__) :
-        sleep(0.5)
+        condition_object.wait(0.5)
 
 
-    logger.debug("exiting main loop")
 
+    logger.debug("exiting main loop")
+    condition_object.release()
     __is_running__ = False
 
 
 
+
 def stop() :
     view.stop()
+
     global __stop_app__
     global __is_running__
+    condition_object.acquire()
+    condition_object.notify()
     __stop_app__ = True
-    logger.info("Application Stopped, now exiting...")
 
+    logger.info("Application Stopped, now exiting...")
+    condition_object.release()
 
 
 def isRunning() :

Backend/Sources/persistence.py

@@ -8,6 +8,8 @@ from utility.app_logging import logger_name
 from sqlalchemy import create_engine
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm.session import sessionmaker
+from sqlalchemy.orm import scoped_session
+from sqlalchemy.pool import SingletonThreadPool
 from sqlalchemy_utils import database_exists, create_database
 
 
@@ -66,7 +68,9 @@ def is_init():
 def init():
     logger.debug("creating db engine....")
     global __db_engine
-    __db_engine = create_engine(__sqlite_db_pattern)
+
+    __db_engine = create_engine(__sqlite_db_pattern,
+                                connect_args={'check_same_thread': False}) # Warning that's not so safe.
 
     # from https://stackoverflow.com/questions/2614984/sqlite-sqlalchemy-how-to-enforce-foreign-keys
     def _fk_pragma_on_connect(dbapi_con, con_record):
@@ -95,10 +99,10 @@ def init():
 
     else:
         logger.error("database is NOT compliant to the model")
-        raise "database is not compliant to the model !"
+        raise Exception("database is not compliant to the model !")
 
     global __db_session
-    __db_session = sessionmaker(bind=__db_engine)
+    __db_session = scoped_session(sessionmaker(bind=__db_engine))
     global __initialized
     __initialized = True
     logger.info("Persistence Session started.")

Backend/Sources/utility/privilege_manager.py

@@ -215,7 +215,7 @@ class Privilege_Manager :
 
         domain = Privilege_Domain(name=name,description=description)
         self.__domains__.add(domain)
-        logger.info("domain with name " + name + "created")
+        logger.info("domain with name " + name + " created")
 
 
     def register_domain(self, domain : Privilege_Domain) :
@@ -225,7 +225,7 @@ class Privilege_Manager :
                 raise BaseException("cannot have two privilege domain with the same name !")
 
         self.__domains__.add(domain)
-        logger.info("domain with name " + domain.name + "created")
+        logger.info("domain with name " + domain.name + " registered")
 
 
 

Backend/Tests/core_test.py

@@ -16,7 +16,7 @@ import Sources.View.view_manager as view
 item_id_to_delete = 0
 import threading
 import pytest
-from pytest import Module, Session
+
 
 
 @pytest.fixture(autouse=True,scope='session')
@@ -53,25 +53,6 @@ def test_login() :
     assert r.status_code == 200
 
 
-def test_inventory() :
-    s = requests.Session()
-
-    r = s.post("http://127.0.0.1:8000/api/login",json={'username' : 'admin','password': 'aseqzdwxc'})
-    print(r.__dict__)
-    assert r.status_code == 200
-
-    r = s.get("http://127.0.0.1:8000/tab")
-    print(r.__dict__)
-    assert r.status_code == 200
-
-    r = s.post("http://127.0.0.1:8000/inventory/item",json={"assign_date_epoch": 5000, "assigned": True, "brand": "Mikrotik", "name": "RB2011", "note": "saline RT 2, chez ashvin", "owned": True, "serial_number": "12345678", "tags": "#router #mikrotik #saline", "type": "router", "wear_score": 100})
-    print(r.__dict__)
-    assert r.status_code == 200
-
-
-    r = s.get("http://127.0.0.1:8000/inventory/items")
-    print(r.__dict__)
-    assert r.status_code == 200
 
 
 

Backend/Tests/inventory_api_test.py

@@ -0,0 +1,90 @@
+
+import os
+from time import sleep
+
+
+os.chdir(os.path.dirname(__file__))
+import sys
+sys.path.append("..")
+sys.path.append("..\\Sources")
+from Sources.Modules.Inventory.inventory import * #altered
+import Sources.main as main_func
+import Sources.Model.model_manager as model_manager
+import requests
+from werkzeug.security import generate_password_hash
+import Sources.View.view_manager as view
+item_id_to_delete = 0
+import threading
+import pytest
+
+
+
+@pytest.fixture(autouse=True,scope='session')
+def fixture2():
+    def run() :
+        main_func.init()
+        persistence.wipeout_database()
+        persistence.init()
+        model_manager.init()
+        main_func.main()
+    t1 = threading.Thread(target=run)
+    t1.start()
+    while(not main_func.isRunning()) :
+        sleep(1)
+
+    yield
+
+    main_func.stop()
+    t1.join()
+
+    # Perform cleanup on the data when the test function exits
+
+def test_api() :
+
+    sleep(2)
+    s = requests.Session()
+
+    print("CREATE")
+    r = s.post("http://127.0.0.1:8000/api/login",json={'username' : 'admin','password': 'aseqzdwxc'})
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("CREATE")
+    r = s.get("http://127.0.0.1:8000/tab")
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("CREATE")
+    r = s.post("http://127.0.0.1:8000/api/inventory/item",json={"assign_date_epoch": 5000, "assigned": True, "brand": "Mikrotik", "name": "RB2011", "note": "saline 2 RT 2, chez ashvin", "owned": True, "serial_number": "12345678", "tags": "#router #mikrotik #saline", "type": "router", "wear_score": 100})
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("CREATE")
+    r = s.post("http://127.0.0.1:8000/api/inventory/item",json={"assign_date_epoch": 10000, "assigned": True, "brand": "Mikrotik", "name": "HEX S 1", "note": "saline 2 RT 3, chez ashvin", "owned": True, "serial_number": "12345678", "tags": "#router #mikrotik #saline", "type": "router", "wear_score": 100})
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("CREATE")
+    r = s.post("http://127.0.0.1:8000/api/inventory/item",json={"assign_date_epoch": 10000, "assigned": True, "brand": "Mikrotik", "name": "HEX S 1", "note": "saline 2 RT 3, chez ashvin", "owned": True, "serial_number": "12345678", "tags": "#router #mikrotik #saline", "type": "router", "wear_score": 100})
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("CREATE")
+    r = s.post("http://127.0.0.1:8000/api/inventory/item",json={"assign_date_epoch": 10000, "assigned": True, "brand": "Mikrotik", "name": "HEX S 1", "note": "saline 2 RT 3, chez ashvin", "owned": True, "serial_number": "12345678", "tags": "#router #mikrotik #saline", "type": "router", "wear_score": 100})
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("GET")
+    r = s.get("http://127.0.0.1:8000/api/inventory/items")
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("DELETE")
+    r = s.delete("http://127.0.0.1:8000/api/inventory/item/2")
+    print(r.__dict__)
+    assert r.status_code == 200
+
+    print("CREATE SITE")
+    r = s.post("http://127.0.0.1:8000/api/inventory/site",json={"assign_date_epoch": 5000, "assigned": True, "brand": "Mikrotik", "name": "RB2011", "note": "saline 2 RT 2, chez ashvin", "owned": True, "serial_number": "12345678", "tags": "#router #mikrotik #saline", "type": "router", "wear_score": 100})
+    print(r.__dict__)
+    assert r.status_code == 200

Backend/Tests/inventory_test.py

@@ -266,3 +266,6 @@ def test_delete_group() :
         if item.id == tmp_id2 or item.id == tmp_id1:
             if item.group_id == tmp_id_gr :
                 raise BaseException("group not deleted")
+
+
+