test_files/db_test.py

from sales_assistant.db.utils.db_connections import validate_config
from sales_assistant.db.utils.db_utils import read_sql, write_to_table
import pandas as pd

# Validate configuration
required_vars = [
    'SSH_HOSTNAME', 'SSH_USERNAME', 'SSH_PEM_CONTENT',
    'MYSQL_HOST', 'MYSQL_USER', 'MYSQL_PASSWORD', 'MYSQL_DB'
]
validate_config_vars = validate_config(required_vars)
print("Config validation:", validate_config_vars)

# Test basic connection
print("\n=== Testing basic connection ===")
test_connection = read_sql("SELECT 1 as test_column;")
print("Connection test result:")
print(test_connection)

# Test reading from products table
print("\n=== Testing products table read ===")
test_product_table = read_sql("SELECT * FROM streamnet.products_list LIMIT 5;")
print("Products table sample:")
print(test_product_table.head())
print(f"Shape: {test_product_table.shape}")

# Test writing a sample DataFrame
print("\n=== Testing DataFrame write (create new table) ===")
sample_data = pd.DataFrame({
    'id': [1, 2, 3],
    'name': ['Test Product 1', 'Test Product 2', 'Test Product 3'],
    'price': [10.99, 20.50, 15.75],
    'description': ['Sample description 1', 'Sample description 2', 'Sample description 3']
})

print("Sample data to write:")
print(sample_data)

# Create new table using convenience function
success = write_to_table(
    df=sample_data, 
    table_name='test_products',
    schema='streamnet',
    if_exists='replace'  # Use replace to create new table
)

if success:
    print("✅ Create table test successful!")
    
    # Test appending more data
    print("\n=== Testing DataFrame append ===")
    additional_data = pd.DataFrame({
        'id': [4, 5],
        'name': ['Test Product 4', 'Test Product 5'],
        'price': [25.00, 30.99],
        'description': ['Sample description 4', 'Sample description 5']
    })
    
    print("Additional data to append:")
    print(additional_data)
    
    append_success = write_to_table(
        df=additional_data,
        table_name='test_products',
        schema='streamnet',
        if_exists='append'
    )
    
    if append_success:
        print("✅ Append test successful!")
        
        # Read back all the data
        print("\n=== Verifying all written data ===")
        read_back = read_sql("SELECT * FROM streamnet.test_products ORDER BY id;")
        print("All data in test table:")
        print(read_back)
        print(f"Total rows: {len(read_back)}")
    else:
        print("❌ Append test failed!")
        
else:
    print("❌ Create table test failed!")