82 lines
No EOL
3 KiB
Python
82 lines
No EOL
3 KiB
Python
"""
|
|
Test script to verify that the 'tags' column exists in the 'podcasts' table.
|
|
This script can be used to test if the migration has been applied correctly.
|
|
"""
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
|
|
def main():
|
|
"""
|
|
Check if the 'tags' column exists in the 'podcasts' table.
|
|
If not, suggest running the migration.
|
|
"""
|
|
print("Testing database schema...")
|
|
|
|
# Find the database file
|
|
db_path = 'instance/podcastrr.db'
|
|
if not os.path.exists(db_path):
|
|
db_path = 'podcastrr.db'
|
|
if not os.path.exists(db_path):
|
|
print("Error: Database file not found.")
|
|
return 1
|
|
|
|
print(f"Using database: {db_path}")
|
|
|
|
# Connect to the database
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
|
|
# Check if the podcasts table exists
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='podcasts'")
|
|
if not cursor.fetchone():
|
|
print("Error: The 'podcasts' table does not exist in the database.")
|
|
print("You may need to initialize the database first with 'python init_db.py'")
|
|
return 1
|
|
|
|
# Check if the tags column exists
|
|
cursor.execute("PRAGMA table_info(podcasts)")
|
|
columns = [column[1] for column in cursor.fetchall()]
|
|
|
|
if 'tags' in columns:
|
|
print("Success: The 'tags' column exists in the 'podcasts' table.")
|
|
print("The migration has been applied correctly.")
|
|
return 0
|
|
else:
|
|
print("Error: The 'tags' column does not exist in the 'podcasts' table.")
|
|
print("You need to run the migration with 'python run_migrations.py'")
|
|
|
|
# Ask if the user wants to run the migration now
|
|
response = input("Do you want to run the migration now? (y/n): ")
|
|
if response.lower() == 'y':
|
|
print("Running migration...")
|
|
import subprocess
|
|
result = subprocess.run([sys.executable, 'run_migrations.py'],
|
|
capture_output=True,
|
|
text=True)
|
|
print(result.stdout)
|
|
|
|
# Check if the migration was successful
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
cursor.execute("PRAGMA table_info(podcasts)")
|
|
columns = [column[1] for column in cursor.fetchall()]
|
|
|
|
if 'tags' in columns:
|
|
print("Success: The migration was applied successfully.")
|
|
return 0
|
|
else:
|
|
print("Error: The migration failed to add the 'tags' column.")
|
|
return 1
|
|
else:
|
|
return 1
|
|
except Exception as e:
|
|
print(f"Error: {str(e)}")
|
|
return 1
|
|
finally:
|
|
if 'conn' in locals():
|
|
conn.close()
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main()) |