Merge pull request #1600 from totvs/release/v1.20.28rc6

Release/v1.20.28rc6

tir/main.py

@@ -22,7 +22,6 @@ class Webapp():
     """
     def __init__(self, config_path="", autostart=True):
         self.__webapp = WebappInternal(config_path, autostart)
-        self.__database = BaseDatabase(config_path, autostart=False)
         self.config = ConfigLoader()
         self.coverage = self.config.coverage
 
@@ -1279,7 +1278,7 @@ def QueryExecute(self, query, database_driver="", dbq_oracle_server="", database
         >>> # Oracle Example:
         >>> self.oHelper.QueryExecute("SELECT * FROM SA1T10", database_driver="Oracle in OraClient19Home1", dbq_oracle_server="Host:Port/oracle instance", database_server="SERVER_NAME", database_name="DATABASE_NAME", database_user="sa", database_password="123456")
         """
-        return self.__database.query_execute(query, database_driver, dbq_oracle_server, database_server, database_port, database_name, database_user, database_password)
+        return self.__webapp.query_execute(query, database_driver, dbq_oracle_server, database_server, database_port, database_name, database_user, database_password)
 
     def GetConfigValue(self, json_key):
         """

tir/technologies/core/base_database.py

@@ -1,17 +1,13 @@
-from tir.technologies.core.base import Base
-from tir.technologies.webapp_internal import WebappInternal
 import pandas as pd
 import pyodbc
 import re
 from tir.technologies.core.logging_config import logger
+from tir.technologies.core.config import ConfigLoader
 
+class BaseDatabase:
 
-class BaseDatabase(Base):
-
-    def __init__(self, config_path="", autostart=True):
-        super().__init__(config_path, autostart=False)
-        self.webapp_internal = WebappInternal(config_path, autostart=False)
-        self.restart_counter = self.webapp_internal.restart_counter
+    def __init__(self):
+        self.config = ConfigLoader()
 
     def odbc_connect(self, database_driver="", dbq_oracle_server="", database_server="", database_port=1521, database_name="", database_user="", database_password=""):
         """
@@ -27,16 +23,10 @@ def odbc_connect(self, database_driver="", dbq_oracle_server="", database_server
         database_password = self.config.database_password if not database_password else database_password
         dbq_oracle_server = self.config.dbq_oracle_server if not dbq_oracle_server else dbq_oracle_server
 
-        self.check_pyodbc_drivers(database_driver)
-
-        try:
-            if dbq_oracle_server:
-                connection = pyodbc.connect(f'DRIVER={database_driver};dbq={dbq_oracle_server};database={database_name};uid={database_user};pwd={database_password}')
-            else:
-                connection = pyodbc.connect(f'DRIVER={database_driver};server={database_server};port={database_port};database={database_name};uid={database_user};pwd={database_password}')
-        except Exception as error:
-            self.webapp_internal.restart_counter = 3
-            self.webapp_internal.log_error(str(error))
+        if dbq_oracle_server:
+            connection = pyodbc.connect(f'DRIVER={database_driver};dbq={dbq_oracle_server};database={database_name};uid={database_user};pwd={database_password}')
+        else:
+            connection = pyodbc.connect(f'DRIVER={database_driver};server={database_server};port={database_port};database={database_name};uid={database_user};pwd={database_password}')
 
         return connection
 
@@ -45,10 +35,7 @@ def test_odbc_connection(self, connection):
         :param connection:
         :return: cursor attribute if connection ok else return False
         """
-        try:
-            return connection.cursor()
-        except:
-            return False
+        return connection.cursor()
 
     def connect_database(self, query="", database_driver="", dbq_oracle_server="", database_server="", database_port=1521, database_name="", database_user="", database_password=""):
 
@@ -74,15 +61,7 @@ def disconnect_database(self, connection):
                 logger().info('DataBase connection stopped')
         else:
             logger().info('DataBase connection already stopped')
-
-    def check_pyodbc_drivers(self, driver_database):
-        if not next(iter(list(
-                filter(lambda x: x == driver_database.lower(), list(map(lambda x: x.lower(), pyodbc.drivers()))))),
-                    None):
-            error_message = f"Driver: '{driver_database}' isn't a valid driver name!"
-            self.webapp_internal.restart_counter = 3
-            self.webapp_internal.log_error(error_message)
-
+
     def query_execute(self, query, database_driver, dbq_oracle_server, database_server, database_port, database_name, database_user, database_password):
         """
         Return a dictionary if the query statement is a SELECT otherwise print a number of row 
@@ -133,16 +112,11 @@ def query_execute(self, query, database_driver, dbq_oracle_server, database_serv
         if re.findall(r'^(SELECT)', query.upper()):
             df = pd.read_sql(sql=query, con=connection)
             return (df.to_dict())
-        elif re.findall(r'^(UPDATE|DELETE|INSERT)', query.upper()):
-            self.cursor_execute(query, connection)
         else:
-            self.webapp_internal.log_error(f"Not a valid query in {query}")
+            self.cursor_execute(query, connection)
 
     def cursor_execute(self, query, connection):
         cursor = connection.cursor()
-        try:
-            rowcount = cursor.execute(query).rowcount
-        except Exception as error:
-            self.webapp_internal.log_error(str(error))
+        rowcount = cursor.execute(query).rowcount
         logger().info(f'{rowcount} row(s) affected')
         connection.commit()

tir/technologies/core/config.py

@@ -8,9 +8,16 @@ class ConfigLoader:
     This class is instantiated to contain all config information used throughout the execution of the methods.
     """
 
+    _instance = None
     _json_data = None
 
-    def __init__(self, path="config.json"):
+    def __new__(cls, path="config.json"):
+        if cls._instance is None:
+            cls._instance = super(ConfigLoader, cls).__new__(cls)
+            cls._instance._initialize(path)
+        return cls._instance
+
+    def _initialize(self, path="config.json"):
         if ConfigLoader._json_data is None:
 
             if not path:
@@ -29,6 +36,8 @@ def __init__(self, path="config.json"):
                         raise Exception(f"JSON file issue: {e}. \n* Please check your config.json *")
 
         if ConfigLoader._json_data:
+            for key, value in ConfigLoader._json_data.items():
+                setattr(self, key, value)
 
             data = ConfigLoader._json_data
 
@@ -61,13 +70,6 @@ def __init__(self, path="config.json"):
             self.skip_restart = ("SkipRestart" in data and bool(data["SkipRestart"]))
             self.smart_test = ("SmartTest" in data and bool(data["SmartTest"]))
             self.smart_erp = ("SmartERP" in data and bool(data["SmartERP"]))
-            self.valid_language = self.language != ""
-            self.initial_program = ""
-            self.routine = ""
-            self.date = ""
-            self.group = ""
-            self.branch = ""
-            self.module = ""
             self.user_cfg = str(data["UserCfg"]) if "UserCfg" in data else ""
             self.password_cfg = str(data["PasswordCfg"]) if "PasswordCfg" in data else ""
             self.electron_binary_path = (str(data["BinPath"]) if "BinPath" in data else "")
@@ -104,6 +106,14 @@ def __init__(self, path="config.json"):
                         "SSLChromeInstallDisable" in data and bool(data["SSLChromeInstallDisable"]))
             self.data_delimiter = str(data["DataDelimiter"]) if "DataDelimiter" in data else "/"
             self.procedure_menu = str(data["ProcedureMenu"]) if "ProcedureMenu" in data else ""
+            self.valid_language = self.language != ""
+            self.initial_program = ""
+            self.routine = ""
+            self.date = ""
+            self.group = ""
+            self.branch = ""
+            self.module = ""
+            self.routine_type = ""
 
     def check_keys(self, json_data):
         valid_keys = [

tir/technologies/poui_internal.py

@@ -93,7 +93,6 @@ def __init__(self, config_path="", autostart=True):
         self.tmenu_screen = None
         self.grid_memo_field = False
         self.range_multiplier = None
-        self.routine = None
 
         if not Base.driver:
             Base.driver = self.driver
@@ -1289,9 +1288,9 @@ def restart(self):
 
 
             if self.config.routine:
-                if self.routine == 'SetLateralMenu':
+                if self.config.routine_type.lower() == 'setlateralmenu':
                     self.SetLateralMenu(self.config.routine, save_input=False)
-                elif self.routine == 'Program':
+                elif self.config.routine_type.lower() == 'program':
                     self.set_program(self.config.routine)
 
     def driver_refresh(self):
@@ -3440,7 +3439,7 @@ def POSearch(self, content, placeholder):
         element().clear()
         element().send_keys(content)
 
-        action = lambda: self.soup_to_selenium(next(iter(input.parent.select('span'))))
+        action = lambda: self.soup_to_selenium(next(iter(po_page.select('po-icon'))))
         ActionChains(self.driver).move_to_element(action()).click().perform()
 
     def ClickTable(self, first_column, second_column, first_content, second_content, table_number, itens, click_cell, checkbox):

tir/technologies/webapp_internal.py

@@ -9,14 +9,16 @@
 import shutil
 import cv2
 import socket
+import pathlib
+import sys
+import tir.technologies.core.enumerations as enum
 from functools import reduce
 from selenium.webdriver.common.keys import Keys
 from bs4 import BeautifulSoup, Tag
 from selenium.webdriver.support import expected_conditions as EC
 from selenium.webdriver.common.by import By
 from selenium.webdriver.common.action_chains import ActionChains
 from selenium.webdriver.support.ui import Select
-import tir.technologies.core.enumerations as enum
 from tir.technologies.core import base
 from tir.technologies.core.log import Log, nump
 from tir.technologies.core.config import ConfigLoader
@@ -29,8 +31,7 @@
 from selenium.common.exceptions import *
 from datetime import datetime
 from tir.technologies.core.logging_config import logger
-import pathlib
-import sys
+from tir.technologies.core.base_database import BaseDatabase
 
 class WebappInternal(Base):
     """
@@ -104,7 +105,6 @@ def __init__(self, config_path="", autostart=True):
         self.tmenu_screen = None
         self.grid_memo_field = False
         self.range_multiplier = None
-        self.routine = None
         self.test_suite = []
         self.current_test_suite = self.log.get_file_name('testsuite')
         self.restart_tss = False
@@ -1561,7 +1561,7 @@ def Program(self, program_name):
         >>> # Calling the method:
         >>> oHelper.Program("MATA020")
         """
-        self.routine = 'Program'
+        self.config.routine_type = 'Program'
         self.config.routine = program_name
 
         if self.config.log_info_config:
@@ -3274,9 +3274,9 @@ def restart(self):
 
 
             if self.config.routine:
-                if self.routine == 'SetLateralMenu':
+                if self.config.routine_type == 'SetLateralMenu':
                     self.SetLateralMenu(self.config.routine, save_input=False)
-                elif self.routine == 'Program':
+                elif self.config.routine_type == 'Program':
                     self.set_program(self.config.routine)
 
     def wait_user_screen(self):
@@ -3964,7 +3964,7 @@ def SetLateralMenu(self, menu_itens, save_input=True, click_menu_functional=Fals
                 self.log_error_newlog()
 
         if save_input:
-            self.routine = 'SetLateralMenu'
+            self.config.routine_type = 'SetLateralMenu'
             self.config.routine = menu_itens
 
         if self.webapp_shadowroot():
@@ -11151,4 +11151,14 @@ def get_container_selector(self, selector):
 
         container = self.get_current_container()
 
-        return container.select(selector)
+        return container.select(selector)
+
+    def query_execute(self, query, database_driver, dbq_oracle_server, database_server, database_port, database_name, database_user, database_password):
+        """
+        Execute a query in a database
+        """
+        base_database = BaseDatabase()
+        try:
+            return base_database.query_execute(query, database_driver, dbq_oracle_server, database_server, database_port, database_name, database_user, database_password)
+        except Exception as e:
+            self.log_error(f"Error in query_execute: {str(e)}")