"""Database level utility class. Used for constructing and seeding the test SQLite database. """ from datetime import datetime import sys import sqlalchemy from asset_transcoder import config from asset_transcoder.connectors import mysql from asset_transcoder.models import asset_final from asset_transcoder.models import asset_status from asset_transcoder.models import asset_upload assets_upload_data = [ dict( id=1, user_id='auth0-test-user-1', asset_type='JPG', filename='bucket_filename_1', object_id='1', object_type='episode', original_filename='episode_photo_1.jpg', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0), ), dict( id=2, user_id='auth0-test-user-1', asset_type='JPG', filename='bucket_filename_2', object_id='1', object_type='podcast', original_filename='podcast_photo_1.jpg', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0) ), dict( id=3, user_id='auth0-test-user-1', asset_type='JPG', filename='bucket_filename_3', object_id='2', object_type='episode', original_filename='episode_photo_3.jpg', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0) ), dict( id=4, user_id='auth0-test-user-1', asset_type='WAV', filename='bucket_filename_4', object_id='2', object_type='episode', original_filename='episode_2_audio.wav', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0) ), dict( id=5, user_id='auth0-test-user-1', asset_type='WAV', filename='bucket_filename_5', object_id='1', object_type='adupload', original_filename='episode_2_audio.wav', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0) ), dict( id=6, user_id='auth0-test-user-1', asset_type='WAV', filename='bucket_filename_6', object_id='3', object_type='adupload', original_filename='episode_2_audio.wav', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0) ), dict( id=7, user_id='auth0-test-user-1', asset_type='WAV', filename='bucket_filename_7', object_id=None, object_type='episode', original_filename='episode_4_audio.wav', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0) ), dict( id=8, user_id='auth0-test-user-1', asset_type='WAV', filename='bucket_filename_8', object_id='4', object_type='episode', original_filename='episode_8_audio.wav', created_date=datetime(2018, 1, 1, 0, 0, 0), updated_date=datetime(2018, 1, 1, 0, 0, 0) ), ] assets_status_data = [ dict( id=1, asset_upload_id='3', status='status 1', created_date=datetime(2018, 1, 1) ), dict( id=2, asset_upload_id='3', status='status 2', created_date=datetime(2019, 1, 1) ), dict( id=3, asset_upload_id='3', status='encoding_completed', created_date=datetime(2019, 1, 1), ), dict( id=4, asset_upload_id='6', status='encoding_error', created_date=datetime(2020, 1, 1), error_code='sample_rate|file_size', error='{"sample_rate": "Invalid input value for sample_rate. Expected value(s): [44100, 48000]", "file_size": ' '"Invalid input value for file_size. Expected value(s): greater than 175212.00000000003"}', ), dict( id=5, asset_upload_id='8', status='encoding_completed', created_date=datetime(2019, 1, 1) ), dict( id=6, asset_upload_id='2', status='encoding_completed', created_date=datetime(2019, 1, 1) ), dict( id=7, asset_upload_id='5', status='acknowledgement_completed', created_date=datetime(2019, 1, 1) ), dict( id=8, asset_upload_id='5', status='validation_error', created_date=datetime(2019, 1, 2) ) ] assets_final_data = [ dict( id=2, asset_upload_id='2', asset_type='JPG', asset_subtype='large_cover', filename='test_bucket_filename_2', duration=900 ), dict( id=3, asset_upload_id='3', asset_type='JPG', asset_subtype='large_cover', bucket='test_bucket', filename='test_bucket_filename_3', duration=0 ), dict( id=4, asset_upload_id='4', asset_type='MP3', asset_subtype=None, filename='test_bucket_filename_4', duration=900 ), dict( id=5, asset_upload_id='5', asset_type='MP3', asset_subtype=None, filename='test_ad_read_filename_1', duration=900 ), dict( id=6, asset_upload_id='3', asset_type='JPG', asset_subtype='xlarge_cover', filename='test_bucket_filename_6', duration=0 ), dict( id=7, asset_upload_id='4', asset_type='FLAC', asset_subtype=None, filename='test_bucket_filename_7', duration=900 ), dict( id=8, asset_upload_id='4', asset_type='DAT', asset_subtype=None, filename='test_bucket_filename_8', duration=0 ) ] def create_tables(): """Create tables defined in asset_transcoder.models.""" if config.ENVIRONMENT != config.TEST_ENVIRONMENT: return if config.TRANSCODER_DB_URL != 'sqlite://': return mysql.BaseModel.metadata.create_all(mysql.asset_transcoder_db_engine) def drop_tables(): """Delete tables defined in test database.""" if config.ENVIRONMENT != config.TEST_ENVIRONMENT: return if config.TRANSCODER_DB_URL != 'sqlite://': return mysql.BaseModel.metadata.drop_all(mysql.asset_transcoder_db_engine) def seed_data(): """Insert test data into test database.""" if config.ENVIRONMENT != config.TEST_ENVIRONMENT: return if config.TRANSCODER_DB_URL != 'sqlite://': return with mysql.transcoder_db_session() as session: try: session.execute(asset_upload.AssetUpload.__table__.insert().values( assets_upload_data )) session.execute(asset_status.AssetStatus.__table__.insert().values( assets_status_data )) session.execute(asset_final.AssetFinal.__table__.insert().values( assets_final_data )) except sqlalchemy.exc.CompileError as error: print('Seeding error: SQL compile:', error) except sqlalchemy.exc.IntegrityError as error: print('Seeding error: Integrity error:', error) except Exception: print('Unexpected error while seeding:', sys.exc_info()[0]) session.commit()