Update, improve and refactor code

.buildkite/node_sync_tests.sh

@@ -39,5 +39,5 @@ node_start_arguments2=${10}
 echo " ==== start sync test"
 python ./sync_tests/node_sync_test.py -b "$build_mode" -e "$env" -t1 "$tag_no1" -t2 "$tag_no2" -r1 "$node_rev1" -r2 "$node_rev2" -n1 "$node_topology1" -n2 "$node_topology2" -a1="$node_start_arguments1" -a2="$node_start_arguments2"
 
-echo " ==== write sync test values into the db"
+echo "--- Prepare for adding sync test results to the AWS Database"
 python ./sync_tests/node_write_sync_values_to_db.py -e "$env"

.buildkite/node_sync_tests.yml

@@ -2,7 +2,6 @@ steps:
   - label: 'run the cardano node sync test on Mainnet using a Linux machine'
     commands:
       - './.buildkite/node_sync_tests.sh $env $build_mode $node_rev1 $node_rev2 $tag_no1 $tag_no2 $node_topology1 $node_topology2 $node_start_arguments1 $node_start_arguments2'
-    branches: "sync_tests"
     timeout_in_minutes: 5000
     agents:
       system: x86_64-linux

.github/workflows/node_sync_test.yaml

@@ -6,9 +6,11 @@ on:
       - sync_tests
     inputs:
       build_mode:
-        description: how to get the cardano node/cli files - nix(, cabal, prebuilt)
-        required: true
-        default: "nix"
+        description: how to get the node & cli executables - currently only nix is supported and windows agents can use only cabal
+        type: choice
+        options: 
+        - nix
+        default: nix
       tag_no1:
         description: rev_label in db/visuals (1.33.0-rc2 (tag number) or 1.33.0 (release number) or 1.33.0_PR2124 (for not released and not tagged runs with a specific node PR/version))
         required: true
@@ -19,8 +21,11 @@ on:
         default: "None"
       node_topology1:
         description: desired cardano-node topology type (used for initial sync) - legacy, p2p
-        required: true
-        default: "legacy"
+        type: choice
+        options: 
+        - legacy
+        - p2p
+        default: legacy
       node_start_arguments1:
         description: extra arguments to be used when starting the node using tag_no1 (--a1 --a2 21)
         required: false
@@ -35,8 +40,11 @@ on:
         default: "None"
       node_topology2:
         description: desired cardano-node topology type (used for final sync) - legacy, p2p
-        required: true
-        default: "legacy"
+        type: choice
+        options: 
+        - legacy
+        - p2p
+        default: legacy
       node_start_arguments2:
         description: extra arguments to be used when starting the node using tag_no2 (--a1 --a2 21)
         required: false

sync_tests/aws_db_utils.py

@@ -3,6 +3,10 @@
 import pymysql.cursors
 import pandas as pd
 
+from utils import sh_colors, print_info, print_ok, print_warn, print_error
+
+
+
 
 def create_connection():
     conn = None
@@ -14,7 +18,7 @@ def create_connection():
                                )
         return conn
     except Exception as e:
-        print(f"!!! Database connection failed due to: {e}")
+        print_error(f"!!! Database connection failed due to: {e}")
 
     return conn
 
@@ -27,7 +31,7 @@ def create_table(table_sql_query):
         conn.commit()
         cur.close()
     except Exception as e:
-        print(f"!!! ERROR: Failed to create table: {e}")
+        print_error(f"!!! ERROR: Failed to create table: {e}")
         return False
     finally:
         if conn:
@@ -39,12 +43,12 @@ def create_table_based_on_another_table_structure(existing_table_name, new_table
 
     conn = create_connection()
     sql_query = f"CREATE TABLE {new_table_name} LIKE {existing_table_name};"
-    print(sql_query)
+    print_info(sql_query)
     try:
         cur = conn.cursor()
         cur.execute(sql_query)
     except Exception as e:
-        print(f"!!! ERROR: Failed create new table {new_table_name} based on {existing_table_name} --> {e}")
+        print_error(f"!!! ERROR: Failed create new table {new_table_name} based on {existing_table_name} --> {e}")
         return False
     finally:
         if conn:
@@ -54,21 +58,22 @@ def create_table_based_on_another_table_structure(existing_table_name, new_table
 def drop_table(table_name):
     conn = create_connection()
     sql_query = f"DROP TABLE {table_name};"
+    print_info(sql_query)
     try:
         cur = conn.cursor()
         cur.execute(sql_query)
         conn.commit()
         cur.close()
     except Exception as e:
-        print(f"!!! ERROR: Failed to drop table {table_name}: {e}")
+        print_error(f"!!! ERROR: Failed to drop table {table_name}: {e}")
         return False
     finally:
         if conn:
             conn.close()
 
 
 def get_column_names_from_table(table_name):
-    print(f"Getting the column names from table: {table_name}")
+    print_info(f"Getting the column names from table: {table_name}")
 
     conn = create_connection()
     sql_query = f"select * from {table_name}"
@@ -79,15 +84,15 @@ def get_column_names_from_table(table_name):
         col_name_list = [res[0] for res in cur.description]
         return col_name_list
     except Exception as e:
-        print(f"!!! ERROR: Failed to get column names from table: {table_name}: {e}")
+        print_error(f"!!! ERROR: Failed to get column names from table: {table_name}: {e}")
         return False
     finally:
         if conn:
             conn.close()
 
 
 def add_column_to_table(table_name, column_name, column_type):
-    print(f"Adding column {column_name} with type {column_type} to {table_name} table")
+    print_info(f"Adding column {column_name} with type {column_type} to {table_name} table")
 
     conn = create_connection()
     sql_query = f"alter table {table_name} add column {column_name} {column_type}"
@@ -96,15 +101,15 @@ def add_column_to_table(table_name, column_name, column_type):
         cur = conn.cursor()
         cur.execute(sql_query)
     except Exception as e:
-        print(f"!!! ERROR: Failed to add {column_name} column into table {table_name} --> {e}")
+        print_error(f"!!! ERROR: Failed to add {column_name} column into table {table_name} --> {e}")
         return False
     finally:
         if conn:
             conn.close()
 
 
 def add_single_value_into_db(table_name, col_names_list, col_values_list):
-    print(f"Adding 1 new entry into {table_name} table")
+    print_info(f"Adding 1 new entry into {table_name} table")
     initial_rows_no = get_last_row_no(table_name)
     col_names = ','.join(col_names_list)
     col_spaces = ','.join(['%s'] * len(col_names_list))
@@ -117,18 +122,23 @@ def add_single_value_into_db(table_name, col_names_list, col_values_list):
         conn.commit()
         cur.close()
     except Exception as e:
-        print(f"  -- !!! ERROR: Failed to insert data into {table_name} table: {e}")
+        print_error(f"  -- !!! ERROR: Failed to insert data into {table_name} table: {e}")
         return False
     finally:
         if conn:
             conn.close()
     final_rows_no = get_last_row_no(table_name)
-    print(f"Successfully added {final_rows_no - initial_rows_no} rows into table {table_name}")
-    return True
+
+    if final_rows_no > 0:
+        print_ok(f"Successfully added {final_rows_no - initial_rows_no} rows into table {table_name}")
+        return True
+
+    print_error(f"Rows were NOT inserted ! final_rows_no: {final_rows_no}")
+    return False
 
 
 def add_bulk_values_into_db(table_name, col_names_list, col_values_list):
-    print(f"Adding {len(col_values_list)} entries into {table_name} table")
+    print_info(f"Adding {len(col_values_list)} entries into {table_name} table")
     initial_rows_no = get_last_row_no(table_name)
     col_names = ','.join(col_names_list)
     col_spaces = ','.join(['%s'] * len(col_names_list))
@@ -141,18 +151,22 @@ def add_bulk_values_into_db(table_name, col_names_list, col_values_list):
         conn.commit()
         cur.close()
     except Exception as e:
-        print(f"  -- !!! ERROR: Failed to bulk insert data into {table_name} table: {e}")
+        print_error(f"  -- !!! ERROR: Failed to bulk insert data into {table_name} table: {e}")
         return False
     finally:
         if conn:
             conn.close()
     final_rows_no = get_last_row_no(table_name)
-    print(f"Successfully added {final_rows_no - initial_rows_no} rows into table {table_name}")
-    return True
+    if final_rows_no > 0:
+        print_ok(f"Successfully added {final_rows_no - initial_rows_no} rows into table {table_name}")
+        return True
+
+    print_error(f"Rows were NOT inserted ! final_rows_no: {final_rows_no}")
+    return False
 
 
 def get_last_row_no(table_name):
-    print(f"Getting the no of rows from table: {table_name}")
+    print_info(f"Getting the no of rows from table: {table_name}")
 
     conn = create_connection()
     sql_query = f"SELECT count(*) FROM {table_name};"
@@ -163,15 +177,15 @@ def get_last_row_no(table_name):
         last_row_no = cur.fetchone()[0]
         return last_row_no
     except Exception as e:
-        print(f"!!! ERROR: Failed to get the no of rows from table {table_name} --> {e}")
+        print_error(f"!!! ERROR: Failed to get the no of rows from table {table_name} --> {e}")
         return False
     finally:
         if conn:
             conn.close()
 
 
 def get_identifier_last_run_from_table(table_name):
-    print(f"Getting the Identifier value of the last run from table {table_name}")
+    print_info(f"Getting the Identifier value of the last run from table {table_name}")
 
     if get_last_row_no(table_name) == 0:
         return table_name + "_0"
@@ -186,15 +200,15 @@ def get_identifier_last_run_from_table(table_name):
             last_identifier = cur.fetchone()[0]
             return last_identifier
         except Exception as e:
-            print(f"!!! ERROR: Failed to get the no of rows from table {table_name} --> {e}")
+            print_error(f"!!! ERROR: Failed to get the no of rows from table {table_name} --> {e}")
             return False
         finally:
             if conn:
                 conn.close()
 
 
 def get_last_epoch_no_from_table(table_name):
-    print(f"Getting the last epoch no value from table {table_name}")
+    print_info(f"Getting the last epoch no value from table {table_name}")
 
     if get_last_row_no(table_name) == 0:
         return 0
@@ -208,15 +222,15 @@ def get_last_epoch_no_from_table(table_name):
             last_identifier = cur.fetchone()[0]
             return last_identifier
         except Exception as e:
-            print(f"!!! ERROR: Failed to get last epoch no from table {table_name} --> {e}")
+            print_error(f"!!! ERROR: Failed to get last epoch no from table {table_name} --> {e}")
             return False
         finally:
             if conn:
                 conn.close()
 
 
 def get_column_values(table_name, column_name):
-    print(f"Getting {column_name} column values from table {table_name}")
+    print_info(f"Getting {column_name} column values from table {table_name}")
 
     conn = create_connection()
     sql_query = f"SELECT {column_name} FROM {table_name};"
@@ -225,15 +239,15 @@ def get_column_values(table_name, column_name):
         cur.execute(sql_query)
         return [el[0] for el in cur.fetchall()]
     except Exception as e:
-        print(f"!!! ERROR: Failed to get {column_name} column values from table {table_name} --> {e}")
+        print_error(f"!!! ERROR: Failed to get {column_name} column values from table {table_name} --> {e}")
         return False
     finally:
         if conn:
             conn.close()
 
 
 def delete_all_rows_from_table(table_name):
-    print(f"Deleting all entries from table: {table_name}")
+    print_info(f"Deleting all entries from table: {table_name}")
     conn = create_connection()
     sql_query = f"TRUNCATE TABLE {table_name}"
     print(f"  -- sql_query: {sql_query}")
@@ -244,7 +258,7 @@ def delete_all_rows_from_table(table_name):
         conn.commit()
         cur.close()
     except Exception as e:
-        print(f"!!! ERROR: Failed to delete all records from table {table_name} --> {e}")
+        print_error(f"!!! ERROR: Failed to delete all records from table {table_name} --> {e}")
         return False
     finally:
         if conn:
@@ -254,9 +268,9 @@ def delete_all_rows_from_table(table_name):
 
 
 def delete_record(table_name, column_name, delete_value):
-    print(f"Deleting rows containing '{delete_value}' value inside the '{column_name}' column")
+    print_info(f"Deleting rows containing '{delete_value}' value inside the '{column_name}' column")
     initial_rows_no = get_last_row_no(table_name)
-    print(f"Deleting {column_name} = {delete_value} from {table_name} table")
+    print_info(f"Deleting {column_name} = {delete_value} from {table_name} table")
 
     conn = create_connection()
     sql_query = f"DELETE from {table_name} where {column_name}=\"{delete_value}\""
@@ -267,7 +281,7 @@ def delete_record(table_name, column_name, delete_value):
         conn.commit()
         cur.close()
     except Exception as e:
-        print(f"!!! ERROR: Failed to delete record {column_name} = {delete_value} from {table_name} table: --> {e}")
+        print_error(f"!!! ERROR: Failed to delete record {column_name} = {delete_value} from {table_name} table: --> {e}")
         return False
     finally:
         if conn:
@@ -277,7 +291,7 @@ def delete_record(table_name, column_name, delete_value):
 
 
 def update_record(table_name, column_name, old_value, new_value):
-    print(f"Updating {column_name} = {new_value} from {table_name} table")
+    print_info(f"Updating {column_name} = {new_value} from {table_name} table")
 
     conn = create_connection()
     sql_query = f"UPDATE {table_name} SET {column_name}=\"{new_value}\" where {column_name}=\"{old_value}\""
@@ -288,7 +302,7 @@ def update_record(table_name, column_name, old_value, new_value):
         conn.commit()
         cur.close()
     except Exception as e:
-        print(f"!!! ERROR: Failed to update record {column_name} = {new_value} from {table_name} table: --> {e}")
+        print_error(f"!!! ERROR: Failed to update record {column_name} = {new_value} from {table_name} table: --> {e}")
         return False
     finally:
         if conn:

sync_tests/gitpython_utils.py

@@ -9,4 +9,5 @@ def git_clone_iohk_repo(repo_name, repo_dir, repo_branch):
 
 
 def git_checkout(repo_name, rev):
+    print(f"Checked out rev: {rev} of {repo_name}")
     repo_name.git.checkout(rev)