Skip to content
Snippets Groups Projects
Select Git revision
  • 34488ce52bca4031a81c57b9b1ee79ce5c4858c6
  • main default protected
  • 75389691-a67c-422a-91e9-aa58bfb5-main-patch-32205
  • test-pipe
  • extended-scripts
  • structured-badges
  • guix-pipeline
  • cabal-pipeline
8 results

CheckMap.hs

Blame
  • check_migrations.py 4.40 KiB
    #!/usr/bin/python3
    import os
    import sys
    import logging
    import datetime
    
    import flask_migrate
    
    from uffd import create_app, db
    from uffd.user.models import User, Group
    from uffd.mfa.models import RecoveryCodeMethod, TOTPMethod, WebauthnMethod
    from uffd.role.models import Role, RoleGroup
    from uffd.signup.models import Signup
    from uffd.invite.models import Invite, InviteGrant, InviteSignup
    from uffd.session.models import DeviceLoginConfirmation
    from uffd.service.models import Service
    from uffd.oauth2.models import OAuth2Client, OAuth2LogoutURI, OAuth2Grant, OAuth2Token, OAuth2DeviceLoginInitiation
    from uffd.selfservice.models import PasswordToken, MailToken
    
    def run_test(dburi, revision):
    	config = {
    		'TESTING': True,
    		'DEBUG': True,
    		'SQLALCHEMY_DATABASE_URI': dburi,
    		'SECRET_KEY': 'DEBUGKEY',
    		'MAIL_SKIP_SEND': True,
    		'SELF_SIGNUP': True,
    		'ENABLE_INVITE': True,
    		'ENABLE_PASSWORDRESET': True,
    		'LDAP_SERVICE_MOCK': True
    	}
    	app = create_app(config)
    	with app.test_request_context():
    		flask_migrate.upgrade(revision='head')
    		# Add a few rows to all tables to make sure that the migrations work with data
    		user = User.query.first()
    		group = Group.query.first()
    		db.session.add(RecoveryCodeMethod(user=user))
    		db.session.add(TOTPMethod(user=user, name='mytotp'))
    		db.session.add(WebauthnMethod(user=user, name='mywebauthn', cred=b''))
    		role = Role(name='role', groups={group: RoleGroup(group=group)})
    		db.session.add(role)
    		role.members.append(user)
    		db.session.add(Role(name='base', included_roles=[role], locked=True, is_default=True, moderator_group=group, groups={group: RoleGroup(group=group)}))
    		db.session.add(Signup(loginname='newuser', displayname='New User', mail='newuser@example.com', password='newpassword'))
    		db.session.add(Signup(loginname='testuser', displayname='Testuser', mail='testuser@example.com', password='testpassword', user=user))
    		invite = Invite(valid_until=datetime.datetime.now(), roles=[role])
    		db.session.add(invite)
    		invite.signups.append(InviteSignup(loginname='newuser', displayname='New User', mail='newuser@example.com', password='newpassword'))
    		invite.grants.append(InviteGrant(user=user))
    		db.session.add(Invite(creator=user, valid_until=datetime.datetime.now()))
    		service = Service(name='testservice', access_group=group)
    		oauth2_client = OAuth2Client(service=service, client_id='testclient', client_secret='testsecret', redirect_uris=['http://localhost:1234/callback'], logout_uris=[OAuth2LogoutURI(method='GET', uri='http://localhost:1234/callback')])
    		db.session.add_all([service, oauth2_client])
    		db.session.add(OAuth2Grant(user=user, client=oauth2_client, code='testcode', redirect_uri='http://example.com/callback', expires=datetime.datetime.now()))
    		db.session.add(OAuth2Token(user=user, client=oauth2_client, token_type='Bearer', access_token='testcode', refresh_token='testcode', expires=datetime.datetime.now()))
    		db.session.add(OAuth2DeviceLoginInitiation(client=oauth2_client, confirmations=[DeviceLoginConfirmation(user=user)]))
    		db.session.add(PasswordToken(user=user))
    		db.session.add(MailToken(user=user, newmail='test@example.com'))
    		db.session.commit()
    		flask_migrate.downgrade(revision=revision)
    		flask_migrate.upgrade(revision='head')
    
    if __name__ == '__main__':
    	if len(sys.argv) != 2 or sys.argv[1] not in ['sqlite', 'mysql']:
    		print('usage: check_migrations.py {sqlite|mysql}')
    		exit(1)
    	dbtype = sys.argv[1]
    	revs = [s.split('_', 1)[0] for s in os.listdir('uffd/migrations/versions') if '_' in s and s.endswith('.py')] + ['base']
    	logging.getLogger().setLevel(logging.INFO)
    	failures = 0
    	for rev in revs:
    		logging.info(f'Testing "upgrade to head, add objects, downgrade to {rev}, upgrade to head"')
    		# Cleanup/drop database
    		if dbtype == 'sqlite':
    			try:
    				os.remove('/tmp/uffd_check_migrations_db.sqlite3')
    			except FileNotFoundError:
    				pass
    			dburi = 'sqlite:////tmp/uffd_check_migrations_db.sqlite3'
    		elif dbtype == 'mysql':
    			import MySQLdb
    			conn = MySQLdb.connect(user='root', unix_socket='/var/run/mysqld/mysqld.sock')
    			cur = conn.cursor()
    			try:
    				cur.execute('DROP DATABASE uffd')
    			except:
    				pass
    			cur.execute('CREATE DATABASE uffd')
    			conn.close()
    			dburi = 'mysql+mysqldb:///uffd?unix_socket=/var/run/mysqld/mysqld.sock'
    		try:
    			run_test(dburi, rev)
    		except Exception as ex:
    			failures += 1
    			logging.error('Test failed', exc_info=ex)
    	if failures:
    		logging.info(f'{failures} tests failed')
    		exit(1)
    	logging.info('All tests succeeded')
    	exit(0)