refac: Update sqlite usages in vulnerability detector lib module #2462

deps/wazuh_testing/wazuh_testing/vulnerability_detector.py

@@ -15,6 +15,7 @@
 from wazuh_testing.tools import file
 from wazuh_testing.tools.services import control_service, check_if_process_is_running
 from wazuh_testing.tools.utils import retry
+from wazuh_testing.tools import sqlite
 from wazuh_testing.wazuh_db import query_wdb
 
 VULN_DETECTOR_GLOBAL_TIMEOUT = 20
@@ -240,60 +241,6 @@ def make_vuln_callback(pattern, prefix=VULNERABILITY_DETECTOR_PREFIX):
  return lambda line: regex.match(line) is not None
 
 
-def load_db(db_path):
- """Load a database in db_path
-
- Args:
- db_path (str): path to the database
- """
- conn = sqlite3.connect(db_path)
- cursor = conn.cursor()
- return conn, cursor
-
-
-def make_query(db_path, query_list):
- """Makes a query to the database in db_path for each item in query_list.
-
- Args:
- db_path (string): Path where is located the DB.
- query_list (list): List with queries to run.
- """
- connect = sqlite3.connect(db_path)
-
- try:
- with connect:
- for item in query_list:
- connect.execute(item)
- finally:
- connect.close()
-
-
-def get_query_result(db_path, query):
- """Return the result of a query in a specified DB
-
- Args:
- db_path (str): path to the database
- query (str): SQL query. (SELECT * ..)
-
- Returns:
- result (List[list]): each row is the query result row and each column is the query field value
- """
- try:
- db, cursor = load_db(db_path)
- cursor.execute(query)
- records = cursor.fetchall()
- result = []
-
- for row in records:
- result.append(', '.join([f"{item}" for item in row]))
-
- return result
-
- finally:
- cursor.close()
- db.close()
-
-
 def clean_table(db_path, table):
  """Deletes all entries in a table of a database.
 
@@ -302,7 +249,7 @@ def clean_table(db_path, table):
  table (str): table from the DB
  """
  query_string = f"DELETE FROM {table}"
- make_query(db_path, [query_string])
+ sqlite.make_query(db_path, [query_string])
 
 
 def clean_table_wazuh_db(agent, table):
@@ -541,7 +488,7 @@ def insert_vulnerability(cveid=DEFAULT_VULNERABILITY_ID, target='RHEL7', target_
  ("{cveid}", "{ref_target}", "{advisory}")
  '''
 
- make_query(CVE_DB_PATH, [vulnerabilities_query_string, query_vulnerabilities_info_query_string,
+ sqlite.make_query(CVE_DB_PATH, [vulnerabilities_query_string, query_vulnerabilities_info_query_string,
  query_references_info, query_bugzilla_info, query_advisories_info])
 
 
@@ -593,7 +540,7 @@ def delete_vulnerability(cveid):
  delete_references_info_query_string = f"DELETE FROM REFERENCES_INFO WHERE id='{cveid}'"
  delete_bugzilla_info_query_string = f"DELETE FROM BUGZILLA_REFERENCES_INFO WHERE id='{cveid}'"
  delete_advisories_info_query_string = f"DELETE FROM ADVISORIES_INFO WHERE id='{cveid}'"
- make_query(CVE_DB_PATH, [delete_vulnerabilities_query_string, delete_vulnerabilities_info_query_string,
+ sqlite.make_query(CVE_DB_PATH, [delete_vulnerabilities_query_string, delete_vulnerabilities_info_query_string,
  delete_references_info_query_string, delete_bugzilla_info_query_string,
  delete_advisories_info_query_string])
 
@@ -639,7 +586,7 @@ def get_num_vulnerabilities():
  int: total number of vulnerabilities in the VULNERABILITIES table
  """
  query_string = f'SELECT count(*) from VULNERABILITIES'
- query_result = get_query_result(CVE_DB_PATH, query_string)
+ query_result = sqlite.get_query_result(CVE_DB_PATH, query_string)
  vulnerabilities_number = int(query_result[0])
 
  return vulnerabilities_number
@@ -872,7 +819,7 @@ def modify_metadata_vuldet_feed(feed, timestamp):
  timestamp (str): The new timestamp value to set.
  """
  query_string = f"update METADATA set TIMESTAMP='{timestamp}' where TARGET='{feed}'"
- make_query(CVE_DB_PATH, [query_string])
+ sqlite.make_query(CVE_DB_PATH, [query_string])
  sleep(1)
 
 
@@ -888,7 +835,7 @@ def modify_nvd_metadata_vuldet(timestamp):
  query_string = f"UPDATE NVD_METADATA SET LAST_UPDATE={timestamp};"
  for _ in range(VULN_DETECTOR_GLOBAL_TIMEOUT):
  try:
- make_query(CVE_DB_PATH, [query_string])
+ sqlite.make_query(CVE_DB_PATH, [query_string])
  break
  except sqlite3.OperationalError:
  sleep(1)