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