change(test_postgresql): success, warning and error print functions

tests/test_postgresql.py

@@ -16,6 +16,36 @@ RED = "\033[91m"
 DEF = "\033[0m"
 
 
+def success(message: str):
+    """Print formatted success message.
+
+    :param message: Message to print
+    :type message: str
+    """
+
+    print(" {}✔{} {}".format(GREEN, DEF, message))
+
+
+def warning(message: str):
+    """Print formatted warning message.
+
+    :param message: Message to print
+    :type message: str
+    """
+
+    print(" {}✔{} {}".format(YELLOW, DEF, message))
+
+
+def error(message: str):
+    """Print formatted error message.
+
+    :param message: Message to print
+    :type message: str
+    """
+
+    print(" {}✔{} {}".format(RED, DEF, message))
+
+
 def is_ha(port: int) -> bool:
     """Check wether this setup is using higlhy-available databases.
 
@@ -234,7 +264,7 @@ def check_write(
             dbname=name, user=user, password=pswd, host=host, port=port
         )
     except psycopg2.Error:
-        print("{}Cannot connect to the database{}".format(RED, DEF))
+        error("Cannot connect to the database")
         return False
 
     # query
@@ -278,7 +308,7 @@ def check_read(
             dbname=name, user=user, password=pswd, host=host, port=port
         )
     except psycopg2.Error:
-        print("{}Cannot connect to the database{}".format(RED, DEF))
+        error("Cannot connect to the database")
         return False
 
     # query
@@ -311,7 +341,7 @@ def check_replication(primary: dict, standby: dict) -> bool:
         primary_client = psycopg2.connect(**primary)
         standby_client = psycopg2.connect(**standby)
     except psycopg2.Error:
-        print("{}Cannot connect to the databases{}".format(RED, DEF))
+        error("Cannot connect to the databases")
         return False
 
     # queries
@@ -379,15 +409,15 @@ def check_ha(db_conn: dict, errors: int = 0, warnings: int = 0) -> tuple:
     # check haproxy
     print("Checking local HAProxy frontends:")
     if not check_listen(db_host, 54321):
-        print("{}✖{} HAProxy pgsql-primary frontend is not listening".format(RED, DEF))
+        error("HAProxy pgsql-primary frontend is not listening")
         errors += 1
     else:
-        print("{}✔{} HAProxy pgsql-primary frontend is listening".format(GREEN, DEF))
+        success("HAProxy pgsql-primary frontend is listening")
     if not check_listen(db_host, 54322):
-        print("{}✖{} HAProxy pgsql-standby frontend is not listening".format(RED, DEF))
+        error("HAProxy pgsql-standby frontend is not listening")
         errors += 1
     else:
-        print("{}✔{} HAProxy pgsql-standby frontend is listening".format(GREEN, DEF))
+        success("HAProxy pgsql-standby frontend is listening")
 
     # check remotes
     print("Checking remote PostgreSQL nodes:")
@@ -395,19 +425,19 @@ def check_ha(db_conn: dict, errors: int = 0, warnings: int = 0) -> tuple:
         node_host = nodes[node]["host"]
         node_port = nodes[node]["port"]
         if not check_listen(node_host, node_port):
-            print("{}✖{} Cannot bind {}:{}".format(RED, DEF, node_host, node_port))
+            error("Cannot bind {}:{}".format(node_host, node_port))
             errors += 1
         else:
-            print("{}✔{} Can bind {}:{}".format(GREEN, DEF, node_host, node_port))
+            success("Can bind {}:{}".format(node_host, node_port))
 
     # check fenced
     print("Checking cluster state:")
     fenced, node = check_fenced(nodes)
     if fenced:
-        print("{}✖{} Node `{}` is fenced".format(RED, DEF, node))
+        error("Node `{}` is fenced".format(node))
         errors += 1
     else:
-        print("{}✔{} No fenced node found".format(GREEN, DEF))
+        success("No fenced node found")
 
     # check replication
     print("Checking replication state:")
@@ -426,10 +456,10 @@ def check_ha(db_conn: dict, errors: int = 0, warnings: int = 0) -> tuple:
         "port": 54322,
     }
     if not check_replication(primary, standby):
-        print("{}✖{} Cannot replicate data between primary/standby".format(RED, DEF))
+        error("Cannot replicate data between primary/standby")
         errors += 1
     else:
-        print("{}✔{} Can replicate data between primary/standby".format(GREEN, DEF))
+        success("Can replicate data between primary/standby")
 
     return errors, warnings
 
@@ -455,42 +485,26 @@ def check_local(db_conn: dict, errors: int = 0, warnings: int = 0) -> tuple:
     # check listen
     print("Checking local PostgreSQL node:")
     if not check_listen(db_host, db_port):
-        print("{}✖{} Cannot connect to {}:{}".format(RED, DEF, db_host, db_port))
+        error("Cannot connect to {}:{}".format(db_host, db_port))
         errors += 1
     else:
-        print("{}✔{} Can connect to {}:{}".format(GREEN, DEF, db_host, db_port))
+        success("Can connect to {}:{}".format(db_host, db_port))
 
     # check read
     print("Checking read operation:")
     if not check_read(db_host, db_port, db_user, db_pass):
-        print(
-            "{}✖{} Cannot read data on {}@{}:{}".format(
-                RED, DEF, db_user, db_host, db_port
-            )
-        )
+        error("Cannot read data on {}@{}:{}".format(db_user, db_host, db_port))
         errors += 1
     else:
-        print(
-            "{}✔{} Can read data on {}@{}:{}".format(
-                GREEN, DEF, db_user, db_host, db_port
-            )
-        )
+        success("Can read data on {}@{}:{}".format(db_user, db_host, db_port))
 
     # check write
     print("Checking write operation:")
     if not check_write(db_host, db_port, db_user, db_pass):
-        print(
-            "{}✖{} Cannot write data on {}@{}:{}".format(
-                RED, DEF, db_user, db_host, db_port
-            )
-        )
+        error("Cannot write data on {}@{}:{}".format(db_user, db_host, db_port))
         errors += 1
     else:
-        print(
-            "{}✔{} Can write data on {}@{}:{}".format(
-                GREEN, DEF, db_user, db_host, db_port
-            )
-        )
+        success("Can write data on {}@{}:{}".format(db_user, db_host, db_port))
 
     return errors, warnings
 
@@ -504,7 +518,7 @@ def main():
 
     # check envsetup utils presence
     if not os.path.isfile(utils):
-        print("{} not found.".format(utils))
+        error("{} not found.".format(utils))
         sys.exit(1)
 
     # load envsetup utils