"""Test database utils module. Used for constructing and seeding the test SQLite database. """ import sys import sqlalchemy from promo_player import config from promo_player.connectors import mysql from promo_player.constants import promo_player as promo_player_constants from promo_player.models import promo_player from promo_player.models import promo_player_track promo_player_data = [ { 'promo_player_id': 1, 'product_id': 2, 'vendor_id': 3, 'code': '0123456789abcdefghij', 'active': True, 'expiration_mode': promo_player_constants.EXPIRATION_MODE_NONE, 'skin': promo_player_constants.SKIN_LIGHT }, { 'promo_player_id': 2, 'product_id': 3, 'vendor_id': 4, 'code': 'abcdefghij0123456789', 'active': True, 'expiration_mode': promo_player_constants.EXPIRATION_MODE_RELEASE, 'skin': promo_player_constants.SKIN_DARK }, { 'promo_player_id': 3, 'product_id': None, 'vendor_id': 5, 'code': 'ababababab1212121212', 'active': True, 'expiration_mode': promo_player_constants.EXPIRATION_MODE_NONE, 'skin': promo_player_constants.SKIN_LIGHT }, { 'promo_player_id': 4, 'product_id': 123, 'vendor_id': 1, 'code': '12341234123412341234', 'active': True, 'expiration_mode': promo_player_constants.EXPIRATION_MODE_NONE, 'skin': promo_player_constants.SKIN_LIGHT }, { 'promo_player_id': 5, 'product_id': 123, 'vendor_id': 1, 'code': '12351235123512351235', 'active': True, 'expiration_mode': promo_player_constants.EXPIRATION_MODE_NONE, 'skin': promo_player_constants.SKIN_LIGHT } ] promo_player_track_data = [ { 'promo_player_id': 1, 'track_id': 1, 'sequence_number': 1 }, { 'promo_player_id': 1, 'track_id': 2, 'sequence_number': 2 }, { 'promo_player_id': 2, 'track_id': 3, 'sequence_number': 1 } ] def in_test_env(): """Check if running the test environment.""" if config.ENVIRONMENT != config.TEST_ENVIRONMENT: return False if config.DB_URL != 'sqlite://': return False return True def create_tables(): """Create tables defined in promo_player.models.""" if not in_test_env(): return mysql.BaseModel.metadata.drop_all(mysql._db_engine) mysql.BaseModel.metadata.create_all(mysql._db_engine) @mysql.autosession() def seed_data(session): """Insert test data.""" if not in_test_env(): return try: session.execute( promo_player.PromoPlayer.__table__.insert().values( promo_player_data)) session.execute( promo_player_track.PromoPlayerTrack.__table__.insert().values( promo_player_track_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()