1from click.testing import CliRunner 2import pytest 3 4from peewee_migrate.cli import cli, get_router 5 6runner = CliRunner() 7 8 9@pytest.fixture 10def dir_option(tmpdir): 11 return '--directory=%s' % tmpdir 12 13 14@pytest.fixture 15def db_url(tmpdir): 16 db_path = '%s/test_sqlite.db' % tmpdir 17 open(db_path, 'a').close() 18 return 'sqlite:///%s' % db_path 19 20 21@pytest.fixture 22def db_option(db_url): 23 return '--database=%s' % db_url 24 25 26@pytest.fixture 27def router(tmpdir, db_url): 28 return lambda: get_router(str(tmpdir), db_url) 29 30 31@pytest.fixture 32def migrations(router): 33 migrations_number = 5 34 name = 'test' 35 for i in range(migrations_number): 36 router().create(name) 37 return ['00%s_test' % i for i in range(1, migrations_number + 1)] 38 39 40@pytest.fixture 41def migrations_str(migrations): 42 return ', '.join(migrations) 43 44 45def test_help(): 46 result = runner.invoke(cli, ['--help']) 47 assert result.exit_code == 0 48 assert 'migrate' in result.output 49 assert 'create' in result.output 50 assert 'rollback' in result.output 51 52 53def test_create(dir_option, db_option): 54 for i in range(2): 55 result = runner.invoke(cli, ['create', dir_option, db_option, '-vvv', 'test']) 56 assert result.exit_code == 0 57 58 59def test_migrate(dir_option, db_option, migrations_str): 60 result = runner.invoke(cli, ['migrate', dir_option, db_option]) 61 assert result.exit_code == 0 62 assert 'Migrations completed: %s' % migrations_str in result.output 63 64 65def test_list(dir_option, db_option, migrations): 66 result = runner.invoke(cli, ['list', dir_option, db_option]) 67 assert 'Migrations are done:\n' in result.output 68 assert 'Migrations are undone:\n%s' % '\n'.join(migrations) in result.output 69 70 71def test_rollback(dir_option, db_option, router, migrations): 72 router().run() 73 74 count_overflow = len(migrations) + 1 75 result = runner.invoke(cli, ['rollback', dir_option, db_option, '--count=%s' % count_overflow]) 76 assert result.exception 77 assert 'Unable to rollback %s migrations' % count_overflow in result.exception.args[0] 78 assert router().done == migrations 79 80 result = runner.invoke(cli, ['rollback', dir_option, db_option]) 81 assert not result.exception 82 assert router().done == migrations[:-1] 83 84 result = runner.invoke(cli, ['rollback', dir_option, db_option, '004_test']) 85 assert not result.exception 86 assert router().done == migrations[:-2] 87 88 result = runner.invoke(cli, ['rollback', dir_option, db_option, '--count=2']) 89 assert not result.exception 90 assert router().done == migrations[:-4] 91 92 result = runner.invoke(cli, ['rollback', dir_option, db_option, '005_test']) 93 assert result.exception 94 assert result.exception.args[0] == 'Only last migration can be canceled.' 95 assert router().done == migrations[:-4] 96 97 98def test_fake(dir_option, db_option, migrations_str, router): 99 result = runner.invoke(cli, ['migrate', dir_option, db_option, '-v', '--fake']) 100 assert result.exit_code == 0 101 assert 'Migrations completed: %s' % migrations_str in result.output 102 103 # TODO: Find a way of testing fake. This is unclear why the following fails. 104 # assert not router().done 105 106 107