Add MySQL integration tests

Author: JustinFrizzell

requirements.txt

@@ -0,0 +1,2 @@
+MYSQL_USERNAME=root
+MYSQL_PASSWORD=MySecretPassw0rd!

tests/integration/inputs/mysql_sqlconnect.env

@@ -0,0 +1,8 @@
+connections:
+  Mysql:
+    sqlalchemy_driver: 'mysql+pymysql'
+    server: 'mysql_db:3306'
+    odbc_driver: ''
+    database: 'sys'
+    username: '${MYSQL_USERNAME}'
+    password: '${MYSQL_PASSWORD}'

tests/integration/inputs/mysql_sqlconnect.yaml

@@ -36,6 +36,7 @@ def create_table_and_insert_data(database_url):
     "postgresql+psycopg2://postgres:mysecretpassword@postgres_db:5432/postgres",
     "mssql+pyodbc://sa:MySecretPassw0rd!@mssql_db:1433/master?driver=ODBC+Driver+17+for+SQL+Server",
     "oracle+oracledb://system:MySecretPassw0rd!@oracle_db:1521/?service_name=XE",
+    "mysql+pymysql://root:MySecretPassw0rd!@mysql_db:3306/sys",
 ]
 
 for url in DATABASE_URLS:

tests/integration/setup/migration.py

@@ -0,0 +1,50 @@
+import os
+from pathlib import Path
+import pytest
+import pandas as pd
+import sqlconnect as sc
+
+if not os.environ.get("RUNNING_IN_DOCKER"):
+    pytest.skip(
+        "Skipping integration tests in local environment", allow_module_level=True
+    )
+
+
+@pytest.fixture(scope="function")
+def setup_env():
+    """Fixture to set up sqlconnect.env"""
+    source_env = Path("tests/integration/inputs/mysql_sqlconnect.env")
+    target_env = Path().home() / "sqlconnect.env"
+
+    target_env.write_text(source_env.read_text(encoding="utf-8"))
+
+    yield
+
+    if target_env.exists():
+        target_env.unlink()
+
+
+@pytest.fixture(scope="function")
+def setup_connections():
+    """Fixture to set up sqlconnect.yaml"""
+    source_env = Path("tests/integration/inputs/mysql_sqlconnect.yaml")
+    target_env = Path().home() / "sqlconnect.yaml"
+
+    target_env.write_text(source_env.read_text(encoding="utf-8"))
+
+    yield
+
+    if target_env.exists():
+        target_env.unlink()
+
+
+def test_sql_to_df_str_postgres(setup_env, setup_connections):
+    conn = sc.Sqlconnector("Mysql")
+
+    df = conn.sql_to_df_str(
+        "SELECT name FROM sys.employees WHERE position = 'Data Engineer'"
+    )
+
+    expected = pd.DataFrame({"name": ["Jane Doe"]})
+
+    pd.testing.assert_frame_equal(df, expected)