mirror/TimeTracker
1
0
Fork 0
mirror of https://github.com/DRYTRIX/TimeTracker.git synced 2026-05-08 05:19:48 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix tests

Browse Source
This commit is contained in:
Dries Peeters
2025-11-01 08:44:02 +01:00
parent 7a28f0665b
commit a110b94a08
6 changed files with 283 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
migrations/versions/037_advanced_expenses.py
+160 -131
View File
@@ -17,134 +17,162 @@ depends_on = None
def upgrade(): def upgrade():
# Create expense_categories table # Import for checking table existence
op.create_table( from sqlalchemy import inspect
'expense_categories',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=100), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('code', sa.String(length=20), nullable=True),
sa.Column('color', sa.String(length=7), nullable=True),
sa.Column('icon', sa.String(length=50), nullable=True),
sa.Column('monthly_budget', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('quarterly_budget', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('yearly_budget', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('budget_threshold_percent', sa.Integer(), nullable=False, server_default='80'),
sa.Column('requires_receipt', sa.Boolean(), nullable=False, server_default='true'),
sa.Column('requires_approval', sa.Boolean(), nullable=False, server_default='true'),
sa.Column('default_tax_rate', sa.Numeric(precision=5, scale=2), nullable=True),
sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('name'),
sa.UniqueConstraint('code')
)
op.create_index('ix_expense_categories_name', 'expense_categories', ['name'], unique=True)
op.create_index('ix_expense_categories_code', 'expense_categories', ['code'], unique=True)
# Create mileage table (without expense_id FK initially) conn = op.get_bind()
op.create_table( inspector = inspect(conn)
'mileage', existing_tables = inspector.get_table_names()
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('project_id', sa.Integer(), nullable=True),
sa.Column('client_id', sa.Integer(), nullable=True),
sa.Column('expense_id', sa.Integer(), nullable=True),
sa.Column('trip_date', sa.Date(), nullable=False),
sa.Column('trip_purpose', sa.Text(), nullable=False),
sa.Column('start_location', sa.String(length=255), nullable=False),
sa.Column('end_location', sa.String(length=255), nullable=False),
sa.Column('distance_km', sa.Numeric(precision=10, scale=2), nullable=False),
sa.Column('vehicle_type', sa.String(length=50), nullable=True),
sa.Column('vehicle_registration', sa.String(length=20), nullable=True),
sa.Column('rate_per_km', sa.Numeric(precision=10, scale=4), nullable=True),
sa.Column('total_amount', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'),
sa.Column('approved_by', sa.Integer(), nullable=True),
sa.Column('approved_at', sa.DateTime(), nullable=True),
sa.Column('rejection_reason', sa.Text(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.PrimaryKeyConstraint('id'),
sa.ForeignKeyConstraint(['user_id'], ['users.id']),
sa.ForeignKeyConstraint(['project_id'], ['projects.id']),
sa.ForeignKeyConstraint(['client_id'], ['clients.id']),
sa.ForeignKeyConstraint(['approved_by'], ['users.id'])
)
op.create_index('ix_mileage_user_id', 'mileage', ['user_id'])
op.create_index('ix_mileage_trip_date', 'mileage', ['trip_date'])
# Create per_diem_rates table # Create expense_categories table (idempotent)
op.create_table( if 'expense_categories' not in existing_tables:
'per_diem_rates', op.create_table(
sa.Column('id', sa.Integer(), nullable=False), 'expense_categories',
sa.Column('country_code', sa.String(length=2), nullable=False), sa.Column('id', sa.Integer(), nullable=False),
sa.Column('location', sa.String(length=255), nullable=True), sa.Column('name', sa.String(length=100), nullable=False),
sa.Column('rate_per_day', sa.Numeric(precision=10, scale=2), nullable=False), sa.Column('description', sa.Text(), nullable=True),
sa.Column('breakfast_deduction', sa.Numeric(precision=10, scale=2), nullable=True), sa.Column('code', sa.String(length=20), nullable=True),
sa.Column('lunch_deduction', sa.Numeric(precision=10, scale=2), nullable=True), sa.Column('color', sa.String(length=7), nullable=True),
sa.Column('dinner_deduction', sa.Numeric(precision=10, scale=2), nullable=True), sa.Column('icon', sa.String(length=50), nullable=True),
sa.Column('valid_from', sa.Date(), nullable=False), sa.Column('monthly_budget', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('valid_to', sa.Date(), nullable=True), sa.Column('quarterly_budget', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('currency_code', sa.String(length=3), nullable=False), sa.Column('yearly_budget', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('notes', sa.Text(), nullable=True), sa.Column('budget_threshold_percent', sa.Integer(), nullable=False, server_default='80'),
sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'), sa.Column('requires_receipt', sa.Boolean(), nullable=False, server_default='true'),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')), sa.Column('requires_approval', sa.Boolean(), nullable=False, server_default='true'),
sa.PrimaryKeyConstraint('id') sa.Column('default_tax_rate', sa.Numeric(precision=5, scale=2), nullable=True),
) sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'),
op.create_index('ix_per_diem_rates_country', 'per_diem_rates', ['country_code']) sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
op.create_index('ix_per_diem_rates_valid_from', 'per_diem_rates', ['valid_from']) sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('name'),
sa.UniqueConstraint('code')
)
op.create_index('ix_expense_categories_name', 'expense_categories', ['name'], unique=True)
op.create_index('ix_expense_categories_code', 'expense_categories', ['code'], unique=True)
# Create per_diems table (without expense_id FK initially) # Create mileage table (without expense_id FK initially) (idempotent)
op.create_table( if 'mileage' not in existing_tables:
'per_diems', op.create_table(
sa.Column('id', sa.Integer(), nullable=False), 'mileage',
sa.Column('user_id', sa.Integer(), nullable=False), sa.Column('id', sa.Integer(), nullable=False),
sa.Column('project_id', sa.Integer(), nullable=True), sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('client_id', sa.Integer(), nullable=True), sa.Column('project_id', sa.Integer(), nullable=True),
sa.Column('expense_id', sa.Integer(), nullable=True), sa.Column('client_id', sa.Integer(), nullable=True),
sa.Column('trip_start_date', sa.Date(), nullable=False), sa.Column('expense_id', sa.Integer(), nullable=True),
sa.Column('trip_end_date', sa.Date(), nullable=False), sa.Column('trip_date', sa.Date(), nullable=False),
sa.Column('destination_country', sa.String(length=2), nullable=False), sa.Column('trip_purpose', sa.Text(), nullable=False),
sa.Column('destination_location', sa.String(length=255), nullable=True), sa.Column('start_location', sa.String(length=255), nullable=False),
sa.Column('per_diem_rate_id', sa.Integer(), nullable=True), sa.Column('end_location', sa.String(length=255), nullable=False),
sa.Column('number_of_days', sa.Integer(), nullable=False), sa.Column('distance_km', sa.Numeric(precision=10, scale=2), nullable=False),
sa.Column('breakfast_provided', sa.Integer(), nullable=False, server_default='0'), sa.Column('vehicle_type', sa.String(length=50), nullable=True),
sa.Column('lunch_provided', sa.Integer(), nullable=False, server_default='0'), sa.Column('vehicle_registration', sa.String(length=20), nullable=True),
sa.Column('dinner_provided', sa.Integer(), nullable=False, server_default='0'), sa.Column('rate_per_km', sa.Numeric(precision=10, scale=4), nullable=True),
sa.Column('total_amount', sa.Numeric(precision=10, scale=2), nullable=True), sa.Column('total_amount', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('currency_code', sa.String(length=3), nullable=False), sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'),
sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'), sa.Column('approved_by', sa.Integer(), nullable=True),
sa.Column('approved_by', sa.Integer(), nullable=True), sa.Column('approved_at', sa.DateTime(), nullable=True),
sa.Column('approved_at', sa.DateTime(), nullable=True), sa.Column('rejection_reason', sa.Text(), nullable=True),
sa.Column('rejection_reason', sa.Text(), nullable=True), sa.Column('notes', sa.Text(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True), sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')), sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')), sa.PrimaryKeyConstraint('id'),
sa.PrimaryKeyConstraint('id'), sa.ForeignKeyConstraint(['user_id'], ['users.id']),
sa.ForeignKeyConstraint(['user_id'], ['users.id']), sa.ForeignKeyConstraint(['project_id'], ['projects.id']),
sa.ForeignKeyConstraint(['project_id'], ['projects.id']), sa.ForeignKeyConstraint(['client_id'], ['clients.id']),
sa.ForeignKeyConstraint(['client_id'], ['clients.id']), sa.ForeignKeyConstraint(['approved_by'], ['users.id'])
sa.ForeignKeyConstraint(['per_diem_rate_id'], ['per_diem_rates.id']), )
sa.ForeignKeyConstraint(['approved_by'], ['users.id']) op.create_index('ix_mileage_user_id', 'mileage', ['user_id'])
) op.create_index('ix_mileage_trip_date', 'mileage', ['trip_date'])
op.create_index('ix_per_diems_user_id', 'per_diems', ['user_id'])
op.create_index('ix_per_diems_trip_start', 'per_diems', ['trip_start_date'])
# Add new columns to expenses table # Create per_diem_rates table (idempotent)
op.add_column('expenses', sa.Column('ocr_data', sa.Text(), nullable=True)) if 'per_diem_rates' not in existing_tables:
op.add_column('expenses', sa.Column('mileage_id', sa.Integer(), nullable=True)) op.create_table(
op.add_column('expenses', sa.Column('per_diem_id', sa.Integer(), nullable=True)) 'per_diem_rates',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('country_code', sa.String(length=2), nullable=False),
sa.Column('location', sa.String(length=255), nullable=True),
sa.Column('rate_per_day', sa.Numeric(precision=10, scale=2), nullable=False),
sa.Column('breakfast_deduction', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('lunch_deduction', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('dinner_deduction', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('valid_from', sa.Date(), nullable=False),
sa.Column('valid_to', sa.Date(), nullable=True),
sa.Column('currency_code', sa.String(length=3), nullable=False),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.PrimaryKeyConstraint('id')
)
op.create_index('ix_per_diem_rates_country', 'per_diem_rates', ['country_code'])
op.create_index('ix_per_diem_rates_valid_from', 'per_diem_rates', ['valid_from'])
# Add foreign keys from expenses to mileage and per_diems # Create per_diems table (without expense_id FK initially) (idempotent)
op.create_foreign_key('fk_expenses_mileage', 'expenses', 'mileage', ['mileage_id'], ['id']) if 'per_diems' not in existing_tables:
op.create_foreign_key('fk_expenses_per_diem', 'expenses', 'per_diems', ['per_diem_id'], ['id']) op.create_table(
'per_diems',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('project_id', sa.Integer(), nullable=True),
sa.Column('client_id', sa.Integer(), nullable=True),
sa.Column('expense_id', sa.Integer(), nullable=True),
sa.Column('trip_start_date', sa.Date(), nullable=False),
sa.Column('trip_end_date', sa.Date(), nullable=False),
sa.Column('destination_country', sa.String(length=2), nullable=False),
sa.Column('destination_location', sa.String(length=255), nullable=True),
sa.Column('per_diem_rate_id', sa.Integer(), nullable=True),
sa.Column('number_of_days', sa.Integer(), nullable=False),
sa.Column('breakfast_provided', sa.Integer(), nullable=False, server_default='0'),
sa.Column('lunch_provided', sa.Integer(), nullable=False, server_default='0'),
sa.Column('dinner_provided', sa.Integer(), nullable=False, server_default='0'),
sa.Column('total_amount', sa.Numeric(precision=10, scale=2), nullable=True),
sa.Column('currency_code', sa.String(length=3), nullable=False),
sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'),
sa.Column('approved_by', sa.Integer(), nullable=True),
sa.Column('approved_at', sa.DateTime(), nullable=True),
sa.Column('rejection_reason', sa.Text(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.Column('updated_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.PrimaryKeyConstraint('id'),
sa.ForeignKeyConstraint(['user_id'], ['users.id']),
sa.ForeignKeyConstraint(['project_id'], ['projects.id']),
sa.ForeignKeyConstraint(['client_id'], ['clients.id']),
sa.ForeignKeyConstraint(['per_diem_rate_id'], ['per_diem_rates.id']),
sa.ForeignKeyConstraint(['approved_by'], ['users.id'])
)
op.create_index('ix_per_diems_user_id', 'per_diems', ['user_id'])
op.create_index('ix_per_diems_trip_start', 'per_diems', ['trip_start_date'])
# Now add the circular foreign keys from mileage and per_diems back to expenses # Add new columns to expenses table (idempotent)
op.create_foreign_key('fk_mileage_expense', 'mileage', 'expenses', ['expense_id'], ['id']) if 'expenses' in existing_tables:
op.create_foreign_key('fk_per_diems_expense', 'per_diems', 'expenses', ['expense_id'], ['id']) existing_columns = [col['name'] for col in inspector.get_columns('expenses')]
if 'ocr_data' not in existing_columns:
op.add_column('expenses', sa.Column('ocr_data', sa.Text(), nullable=True))
if 'mileage_id' not in existing_columns:
op.add_column('expenses', sa.Column('mileage_id', sa.Integer(), nullable=True))
if 'per_diem_id' not in existing_columns:
op.add_column('expenses', sa.Column('per_diem_id', sa.Integer(), nullable=True))
# Add foreign keys from expenses to mileage and per_diems (idempotent)
existing_fks = [fk['name'] for fk in inspector.get_foreign_keys('expenses')]
if 'fk_expenses_mileage' not in existing_fks:
op.create_foreign_key('fk_expenses_mileage', 'expenses', 'mileage', ['mileage_id'], ['id'])
if 'fk_expenses_per_diem' not in existing_fks:
op.create_foreign_key('fk_expenses_per_diem', 'expenses', 'per_diems', ['per_diem_id'], ['id'])
# Now add the circular foreign keys from mileage and per_diems back to expenses (idempotent)
if 'mileage' in existing_tables:
mileage_fks = [fk['name'] for fk in inspector.get_foreign_keys('mileage')]
if 'fk_mileage_expense' not in mileage_fks:
op.create_foreign_key('fk_mileage_expense', 'mileage', 'expenses', ['expense_id'], ['id'])
if 'per_diems' in existing_tables:
per_diems_fks = [fk['name'] for fk in inspector.get_foreign_keys('per_diems')]
if 'fk_per_diems_expense' not in per_diems_fks:
op.create_foreign_key('fk_per_diems_expense', 'per_diems', 'expenses', ['expense_id'], ['id'])
# Insert default expense categories # Insert default expense categories
op.execute(""" op.execute("""
@@ -160,15 +188,16 @@ def upgrade():
ON CONFLICT (name) DO NOTHING ON CONFLICT (name) DO NOTHING
""") """)
# Insert default per diem rates # Insert default per diem rates (idempotent)
op.execute(""" if 'per_diem_rates' in existing_tables:
INSERT INTO per_diem_rates (country_code, location, rate_per_day, breakfast_deduction, lunch_deduction, dinner_deduction, valid_from, currency_code, is_active) op.execute("""
VALUES INSERT OR IGNORE INTO per_diem_rates (country_code, location, rate_per_day, breakfast_deduction, lunch_deduction, dinner_deduction, valid_from, currency_code, is_active)
('US', 'General', 55.00, 13.00, 16.00, 26.00, '2025-01-01', 'USD', true), VALUES
('GB', 'General', 45.00, 10.00, 13.00, 22.00, '2025-01-01', 'GBP', true), ('US', 'General', 55.00, 13.00, 16.00, 26.00, '2025-01-01', 'USD', true),
('DE', 'General', 24.00, 5.00, 8.00, 11.00, '2025-01-01', 'EUR', true), ('GB', 'General', 45.00, 10.00, 13.00, 22.00, '2025-01-01', 'GBP', true),
('FR', 'General', 20.00, 4.00, 7.00, 9.00, '2025-01-01', 'EUR', true) ('DE', 'General', 24.00, 5.00, 8.00, 11.00, '2025-01-01', 'EUR', true),
""") ('FR', 'General', 20.00, 4.00, 7.00, 9.00, '2025-01-01', 'EUR', true)
""")
def downgrade(): def downgrade():
migrations/versions/039_add_budget_alerts_table.py
+33 -26
View File
@@ -17,33 +17,40 @@ depends_on = None
def upgrade(): def upgrade():
"""Create budget_alerts table""" """Create budget_alerts table (idempotent)"""
op.create_table( from sqlalchemy import inspect
'budget_alerts',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('project_id', sa.Integer(), nullable=False),
sa.Column('alert_type', sa.String(length=20), nullable=False),
sa.Column('alert_level', sa.String(length=20), nullable=False),
sa.Column('budget_consumed_percent', sa.Numeric(precision=5, scale=2), nullable=False),
sa.Column('budget_amount', sa.Numeric(precision=10, scale=2), nullable=False),
sa.Column('consumed_amount', sa.Numeric(precision=10, scale=2), nullable=False),
sa.Column('message', sa.Text(), nullable=False),
sa.Column('is_acknowledged', sa.Boolean(), nullable=False, server_default='0'),
sa.Column('acknowledged_by', sa.Integer(), nullable=True),
sa.Column('acknowledged_at', sa.DateTime(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['project_id'], ['projects.id'], name='fk_budget_alerts_project_id', ondelete='CASCADE'),
sa.ForeignKeyConstraint(['acknowledged_by'], ['users.id'], name='fk_budget_alerts_acknowledged_by', ondelete='SET NULL'),
sa.PrimaryKeyConstraint('id')
)
# Create indexes for better query performance conn = op.get_bind()
with op.batch_alter_table('budget_alerts', schema=None) as batch_op: inspector = inspect(conn)
batch_op.create_index('ix_budget_alerts_project_id', ['project_id']) existing_tables = inspector.get_table_names()
batch_op.create_index('ix_budget_alerts_acknowledged_by', ['acknowledged_by'])
batch_op.create_index('ix_budget_alerts_created_at', ['created_at']) if 'budget_alerts' not in existing_tables:
batch_op.create_index('ix_budget_alerts_is_acknowledged', ['is_acknowledged']) op.create_table(
batch_op.create_index('ix_budget_alerts_alert_type', ['alert_type']) 'budget_alerts',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('project_id', sa.Integer(), nullable=False),
sa.Column('alert_type', sa.String(length=20), nullable=False),
sa.Column('alert_level', sa.String(length=20), nullable=False),
sa.Column('budget_consumed_percent', sa.Numeric(precision=5, scale=2), nullable=False),
sa.Column('budget_amount', sa.Numeric(precision=10, scale=2), nullable=False),
sa.Column('consumed_amount', sa.Numeric(precision=10, scale=2), nullable=False),
sa.Column('message', sa.Text(), nullable=False),
sa.Column('is_acknowledged', sa.Boolean(), nullable=False, server_default='0'),
sa.Column('acknowledged_by', sa.Integer(), nullable=True),
sa.Column('acknowledged_at', sa.DateTime(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['project_id'], ['projects.id'], name='fk_budget_alerts_project_id', ondelete='CASCADE'),
sa.ForeignKeyConstraint(['acknowledged_by'], ['users.id'], name='fk_budget_alerts_acknowledged_by', ondelete='SET NULL'),
sa.PrimaryKeyConstraint('id')
)
# Create indexes for better query performance
with op.batch_alter_table('budget_alerts', schema=None) as batch_op:
batch_op.create_index('ix_budget_alerts_project_id', ['project_id'])
batch_op.create_index('ix_budget_alerts_acknowledged_by', ['acknowledged_by'])
batch_op.create_index('ix_budget_alerts_created_at', ['created_at'])
batch_op.create_index('ix_budget_alerts_is_acknowledged', ['is_acknowledged'])
batch_op.create_index('ix_budget_alerts_alert_type', ['alert_type'])
def downgrade(): def downgrade():
migrations/versions/040_add_import_export_tables.py
+54 -47
View File
@@ -18,57 +18,64 @@ depends_on = None
def upgrade(): def upgrade():
"""Create import/export tracking tables""" """Create import/export tracking tables (idempotent)"""
from sqlalchemy import inspect
conn = op.get_bind()
inspector = inspect(conn)
existing_tables = inspector.get_table_names()
# Create data_imports table # Create data_imports table
op.create_table( if 'data_imports' not in existing_tables:
'data_imports', op.create_table(
sa.Column('id', sa.Integer(), nullable=False), 'data_imports',
sa.Column('user_id', sa.Integer(), nullable=False), sa.Column('id', sa.Integer(), nullable=False),
sa.Column('import_type', sa.String(length=50), nullable=False), sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('source_file', sa.String(length=500), nullable=True), sa.Column('import_type', sa.String(length=50), nullable=False),
sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'), sa.Column('source_file', sa.String(length=500), nullable=True),
sa.Column('total_records', sa.Integer(), nullable=False, server_default='0'), sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'),
sa.Column('successful_records', sa.Integer(), nullable=False, server_default='0'), sa.Column('total_records', sa.Integer(), nullable=False, server_default='0'),
sa.Column('failed_records', sa.Integer(), nullable=False, server_default='0'), sa.Column('successful_records', sa.Integer(), nullable=False, server_default='0'),
sa.Column('error_log', sa.Text(), nullable=True), sa.Column('failed_records', sa.Integer(), nullable=False, server_default='0'),
sa.Column('import_summary', sa.Text(), nullable=True), sa.Column('error_log', sa.Text(), nullable=True),
sa.Column('started_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')), sa.Column('import_summary', sa.Text(), nullable=True),
sa.Column('completed_at', sa.DateTime(), nullable=True), sa.Column('started_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ), sa.Column('completed_at', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id') sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
) sa.PrimaryKeyConstraint('id')
)
# Create indexes for data_imports
op.create_index('ix_data_imports_user_id', 'data_imports', ['user_id']) # Create indexes for data_imports
op.create_index('ix_data_imports_status', 'data_imports', ['status']) op.create_index('ix_data_imports_user_id', 'data_imports', ['user_id'])
op.create_index('ix_data_imports_started_at', 'data_imports', ['started_at']) op.create_index('ix_data_imports_status', 'data_imports', ['status'])
op.create_index('ix_data_imports_started_at', 'data_imports', ['started_at'])
# Create data_exports table # Create data_exports table
op.create_table( if 'data_exports' not in existing_tables:
'data_exports', op.create_table(
sa.Column('id', sa.Integer(), nullable=False), 'data_exports',
sa.Column('user_id', sa.Integer(), nullable=False), sa.Column('id', sa.Integer(), nullable=False),
sa.Column('export_type', sa.String(length=50), nullable=False), sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('export_format', sa.String(length=20), nullable=False), sa.Column('export_type', sa.String(length=50), nullable=False),
sa.Column('file_path', sa.String(length=500), nullable=True), sa.Column('export_format', sa.String(length=20), nullable=False),
sa.Column('file_size', sa.Integer(), nullable=True), sa.Column('file_path', sa.String(length=500), nullable=True),
sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'), sa.Column('file_size', sa.Integer(), nullable=True),
sa.Column('filters', sa.Text(), nullable=True), sa.Column('status', sa.String(length=20), nullable=False, server_default='pending'),
sa.Column('record_count', sa.Integer(), nullable=False, server_default='0'), sa.Column('filters', sa.Text(), nullable=True),
sa.Column('error_message', sa.Text(), nullable=True), sa.Column('record_count', sa.Integer(), nullable=False, server_default='0'),
sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')), sa.Column('error_message', sa.Text(), nullable=True),
sa.Column('completed_at', sa.DateTime(), nullable=True), sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),
sa.Column('expires_at', sa.DateTime(), nullable=True), sa.Column('completed_at', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ), sa.Column('expires_at', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id') sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
) sa.PrimaryKeyConstraint('id')
)
# Create indexes for data_exports
op.create_index('ix_data_exports_user_id', 'data_exports', ['user_id']) # Create indexes for data_exports
op.create_index('ix_data_exports_status', 'data_exports', ['status']) op.create_index('ix_data_exports_user_id', 'data_exports', ['user_id'])
op.create_index('ix_data_exports_created_at', 'data_exports', ['created_at']) op.create_index('ix_data_exports_status', 'data_exports', ['status'])
op.create_index('ix_data_exports_expires_at', 'data_exports', ['expires_at']) op.create_index('ix_data_exports_created_at', 'data_exports', ['created_at'])
op.create_index('ix_data_exports_expires_at', 'data_exports', ['expires_at'])
def downgrade(): def downgrade():
tests/test_budget_alert_model.py
+1 -1
View File
@@ -10,7 +10,7 @@ from app.models import BudgetAlert, Project, User, Client
@pytest.fixture @pytest.fixture
def client_obj(app): def client_obj(app):
"""Create a test client""" """Create a test client"""
client = Client(name="Test Client", status="active") client = Client(name="Test Client")
db.session.add(client) db.session.add(client)
db.session.commit() db.session.commit()
return client return client
tests/test_budget_alerts_smoke.py
+1 -1
View File
@@ -30,7 +30,7 @@ def regular_user(app):
@pytest.fixture @pytest.fixture
def client_obj(app): def client_obj(app):
"""Create a test client""" """Create a test client"""
client = Client(name="Smoke Test Client", status="active") client = Client(name="Smoke Test Client")
db.session.add(client) db.session.add(client)
db.session.commit() db.session.commit()
return client return client
tests/test_i18n.py
+34 -15
View File
@@ -63,7 +63,8 @@ class TestLocaleSelection:
db.session.commit() db.session.commit()
# Login as user # Login as user
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Check that locale is set to user's preference # Check that locale is set to user's preference
with client.application.test_request_context(): with client.application.test_request_context():
@@ -105,7 +106,8 @@ class TestLanguageSwitching:
def test_set_language_direct_route(self, client, test_user): def test_set_language_direct_route(self, client, test_user):
"""Test direct language switching route""" """Test direct language switching route"""
# Login first # Login first
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Switch to Spanish # Switch to Spanish
response = client.get('/set-language/es', follow_redirects=False) response = client.get('/set-language/es', follow_redirects=False)
@@ -124,7 +126,8 @@ class TestLanguageSwitching:
def test_set_language_api_endpoint(self, client, test_user): def test_set_language_api_endpoint(self, client, test_user):
"""Test API endpoint for language switching""" """Test API endpoint for language switching"""
# Login first # Login first
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Switch to Arabic via API # Switch to Arabic via API
response = client.post( response = client.post(
@@ -145,7 +148,8 @@ class TestLanguageSwitching:
def test_set_invalid_language(self, client, test_user): def test_set_invalid_language(self, client, test_user):
"""Test that invalid languages are rejected""" """Test that invalid languages are rejected"""
# Login first # Login first
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Try to set invalid language # Try to set invalid language
response = client.post( response = client.post(
@@ -161,14 +165,16 @@ class TestLanguageSwitching:
def test_language_persists_across_sessions(self, client, test_user): def test_language_persists_across_sessions(self, client, test_user):
"""Test that language preference persists across sessions""" """Test that language preference persists across sessions"""
# Login and set language # Login and set language
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
client.get('/set-language/de', follow_redirects=True) client.get('/set-language/de', follow_redirects=True)
# Logout # Logout
client.get('/auth/logout', follow_redirects=True) client.get('/auth/logout', follow_redirects=True)
# Login again # Login again
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Check that language preference is still set # Check that language preference is still set
db.session.refresh(test_user) db.session.refresh(test_user)
@@ -185,12 +191,14 @@ class TestRTLSupport:
db.session.commit() db.session.commit()
# Login # Login
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Get dashboard # Get dashboard
response = client.get('/dashboard') response = client.get('/dashboard')
# Check that page includes RTL directive # Check that page includes RTL directive
assert response.status_code == 200
assert b'dir="rtl"' in response.data or b"dir='rtl'" in response.data assert b'dir="rtl"' in response.data or b"dir='rtl'" in response.data
def test_rtl_detection_for_hebrew(self, client, test_user): def test_rtl_detection_for_hebrew(self, client, test_user):
@@ -200,12 +208,14 @@ class TestRTLSupport:
db.session.commit() db.session.commit()
# Login # Login
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Get dashboard # Get dashboard
response = client.get('/dashboard') response = client.get('/dashboard')
# Check that page includes RTL directive # Check that page includes RTL directive
assert response.status_code == 200
assert b'dir="rtl"' in response.data or b"dir='rtl'" in response.data assert b'dir="rtl"' in response.data or b"dir='rtl'" in response.data
def test_ltr_for_english(self, client, test_user): def test_ltr_for_english(self, client, test_user):
@@ -215,12 +225,14 @@ class TestRTLSupport:
db.session.commit() db.session.commit()
# Login # Login
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Get dashboard # Get dashboard
response = client.get('/dashboard') response = client.get('/dashboard')
# Check that page includes LTR directive # Check that page includes LTR directive
assert response.status_code == 200
assert b'dir="ltr"' in response.data or b"dir='ltr'" in response.data assert b'dir="ltr"' in response.data or b"dir='ltr'" in response.data
@@ -254,22 +266,26 @@ class TestLanguageSelectorUI:
def test_language_selector_in_header(self, client, test_user): def test_language_selector_in_header(self, client, test_user):
"""Test that language selector appears in header""" """Test that language selector appears in header"""
# Login # Login
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Get dashboard # Get dashboard
response = client.get('/dashboard') response = client.get('/dashboard')
# Check that language selector is present # Check that language selector is present
assert response.status_code == 200
assert b'langDropdown' in response.data or b'lang-dropdown' in response.data.lower() assert b'langDropdown' in response.data or b'lang-dropdown' in response.data.lower()
assert b'fa-globe' in response.data or b'globe' in response.data.lower() assert b'fa-globe' in response.data or b'globe' in response.data.lower()
def test_language_list_contains_all_languages(self, client, test_user): def test_language_list_contains_all_languages(self, client, test_user):
"""Test that language selector contains all available languages""" """Test that language selector contains all available languages"""
# Login # Login
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Get dashboard # Get dashboard
response = client.get('/dashboard') response = client.get('/dashboard')
assert response.status_code == 200
response_text = response.data.decode('utf-8') response_text = response.data.decode('utf-8')
# Check for language names in the page # Check for language names in the page
@@ -284,7 +300,8 @@ class TestUserSettingsLanguage:
def test_language_setting_in_user_settings(self, client, test_user): def test_language_setting_in_user_settings(self, client, test_user):
"""Test that language setting is available in user settings""" """Test that language setting is available in user settings"""
# Login # Login
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Get settings page # Get settings page
response = client.get('/settings') response = client.get('/settings')
@@ -296,7 +313,8 @@ class TestUserSettingsLanguage:
def test_save_language_in_user_settings(self, client, test_user): def test_save_language_in_user_settings(self, client, test_user):
"""Test saving language preference in user settings""" """Test saving language preference in user settings"""
# Login # Login
client.post('/auth/login', data={'username': test_user.username}, follow_redirects=True) with client.session_transaction() as sess:
sess['_user_id'] = str(test_user.id)
# Update settings with language # Update settings with language
response = client.post('/settings', data={ response = client.post('/settings', data={
@@ -306,6 +324,7 @@ class TestUserSettingsLanguage:
}, follow_redirects=True) }, follow_redirects=True)
# Check that setting was saved # Check that setting was saved
assert response.status_code == 200
db.session.refresh(test_user) db.session.refresh(test_user)
assert test_user.preferred_language == 'fr' assert test_user.preferred_language == 'fr'
@@ -314,8 +333,8 @@ class TestUserSettingsLanguage:
def test_user(client): def test_user(client):
"""Create a test user""" """Create a test user"""
with client.application.app_context(): with client.application.app_context():
user = User(username='testuser') user = User(username='testuser', role='user')
user.role = 'user' user.is_active = True
db.session.add(user) db.session.add(user)
db.session.commit() db.session.commit()
yield user yield user