Merge pull request fangci221#2 from Hank8933/origin

test: Add test for MySQL server

.gitignore

@@ -170,3 +170,5 @@ output
 
 /green-harrison/src/source/tool-chain/tools.mk
 >>>>>>> b1d0251962a764da3029a4f964128e7288272946
+.coverage
+*.pyc

requirements.txt

@@ -0,0 +1,5 @@
+pytest
+pytest-playwright
+ymysql
+cryptography
+python-dotenv

source/test/database.py

@@ -0,0 +1,33 @@
+import os
+import pymysql
+from dotenv import load_dotenv
+
+
+def connect_mysql():
+
+    # Load environment variables from .env
+    load_dotenv()
+
+    # Set connection parameters
+    config = {
+        'user': os.getenv('DB_USER'),
+        'passwd': os.getenv('DB_PASSWD'),
+        'host': os.getenv('DB_HOST'),
+        'port': int(os.getenv('DB_PORT')),
+        'db': os.getenv('DB_NAME')
+    }
+
+    # Connect to the database
+    try:
+        cnx = pymysql.connect(**config)
+    except pymysql.err.OperationalError as e:
+        print(f"Can't connect to MySQL database: {e}")
+        return None
+    else:
+        return cnx
+
+
+if __name__ == "__main__":
+    cnx = connect_mysql()
+    print("Connection is OK")
+    cnx.close()

source/test/test_import_sql.py

@@ -0,0 +1,52 @@
+import pytest
+import database
+
+
+class TestMySQL:
+
+    # Define a fixture to import the .sql file
+    @pytest.fixture(scope="class")
+    def setup(self):
+
+        # Connect to the database
+        cnx = database.connect_mysql()
+        assert cnx is not None, "Connection is OK"
+
+        # Create a cursor object, which is used to execute SQL statements
+        cursor = cnx.cursor()
+
+        # Open the .sql file and read the content
+        with open("./Dump20231016.sql", "r") as f:
+            sql_content = f.read()
+            sql_list = sql_content.split(";")
+
+        # Execute the SQL statements
+        for sql in sql_list:
+            # If the SQL statement is not empty, execute it
+            if sql.strip():
+                cursor.execute(sql)
+                # If the SQL statement is a CREATE statement, commit the changes
+                if sql.startswith("CREATE"):
+                    cnx.commit()
+
+        yield cnx, cursor
+
+        cursor.close()
+        cnx.close()
+
+
+    # Test if the database is created
+    def test_import_sql(self, setup):
+
+        # Get the cursor object
+        cnx, cursor = setup
+
+        # Check if the database is created
+        cursor.execute("SHOW TABLES")
+        tables = cursor.fetchall()
+        assert len(tables) == 5, "The number of tables is correct"
+
+        # Check if the table "account" is created
+        cursor.execute("SELECT * FROM account")
+        accounts = cursor.fetchall()
+        assert len(accounts) == 2, "The number of accounts is correct"

source/test/test_mysql_login.py

@@ -0,0 +1,10 @@
+import database
+
+
+def test_mysql_login():
+
+    # Connect to the database
+    cnx = database.connect_mysql()
+
+    # Check if the connection is successful
+    assert cnx is not None, "Connection is OK"