ellar-sql/tests/test_migrations/test_migrations_commands.py at 370134eb7138fbe2cc56bf0d3bedef4e40a31815 · python-ellar/ellar-sql · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
from ..utils import clean_directory, run_command, set_env_variable


@clean_directory("default")
def test_migrate_upgrade():
    result = run_command("default.py db init")
    assert result.returncode == 0
    assert b"default/migrations/alembic.ini before proceeding." in result.stdout

    result = run_command("default.py db check")
    assert result.returncode == 1

    result = run_command("default.py db migrate")
    assert result.returncode == 0

    result = run_command("default.py db check")
    assert result.returncode == 1

    result = run_command("default.py db upgrade")
    assert result.returncode == 0

    result = run_command("default.py db check")
    assert result.returncode == 0
    assert result.stdout == b"No new upgrade operations detected.\n"

    result = run_command("default.py add-user")
    assert result.returncode == 0
    assert result.stdout == b"<User name=default App Ellar id=1>\n"


@clean_directory("custom_directory")
def test_migrate_upgrade_custom_directory():
    result = run_command("custom_directory.py db init")
    assert result.returncode == 0
    assert (
        b"custom_directory/temp_migrations/alembic.ini before proceeding."
        in result.stdout
    )

    result = run_command("custom_directory.py db check")
    assert result.returncode == 1

    result = run_command("custom_directory.py db migrate")
    assert result.returncode == 0

    result = run_command("custom_directory.py db check")
    assert result.returncode == 1

    result = run_command("custom_directory.py db upgrade")
    assert result.returncode == 0

    result = run_command("custom_directory.py db check")
    assert result.returncode == 0
    assert result.stdout == b"No new upgrade operations detected.\n"

    result = run_command("custom_directory.py add-user")
    assert result.returncode == 0
    assert result.stdout == b"<User name=Custom Directory App Ellar id=1>\n"


@clean_directory("custom_directory")
def test_migrate_upgrade_custom_directory_with_model_changes():
    result = run_command("custom_directory.py db init")
    assert result.returncode == 0

    result = run_command("custom_directory.py db migrate")
    assert result.returncode == 0

    result = run_command("custom_directory.py db upgrade")
    assert result.returncode == 0

    with set_env_variable("model_change_name", "true"):
        result = run_command("custom_directory.py db migrate")
        assert result.returncode == 0
        assert (
            b"Detected type change from VARCHAR(length=256) to String(length=128)"
            in result.stderr
        )


@clean_directory("default")
def test_other_alembic_commands():
    result = run_command("default.py db init")
    assert result.returncode == 0

    # Revision Command
    result = run_command("default.py db revision")
    assert result.returncode == 0

    # Edit Command
    result = run_command("default.py db edit")
    assert result.returncode == 1
    assert b"Error: Error executing editor" in result.stderr

    # Merge Command
    result = run_command("default.py db merge")
    assert result.returncode == 2
    assert b"Missing argument '<revisions>, ..." in result.stderr

    # Show Command
    result = run_command("default.py db show")
    assert result.returncode == 0

    # History Command
    result = run_command("default.py db history")
    assert result.returncode == 0

    # Heads Command
    result = run_command("default.py db heads")
    assert result.returncode == 0

    # Branches Command
    result = run_command("default.py db branches")
    assert result.returncode == 0

    # Current Command
    result = run_command("default.py db current")
    assert result.returncode == 0

    # Stamp Command
    result = run_command("default.py db stamp")
    assert result.returncode == 1
    assert (
        b"revision identifier False is not a string; ensure database driver settings are correct"
        in result.stderr
    )

    # Downgrade Command
    result = run_command("default.py db downgrade")
    assert result.returncode == 1
    assert b"Relative revision -1 didn't produce 1 migrations" in result.stderr


@clean_directory("default_async")
def test_migrate_upgrade_async():
    result = run_command("default_async.py db init")
    assert result.returncode == 0
    assert b"default_async/migrations/alembic.ini before proceeding." in result.stdout

    result = run_command("default_async.py db check")
    assert result.returncode == 1

    result = run_command("default_async.py db migrate")
    assert result.returncode == 0

    result = run_command("default_async.py db check")
    assert result.returncode == 1

    result = run_command("default_async.py db upgrade")
    assert result.returncode == 0

    result = run_command("default_async.py db check")
    assert result.returncode == 0
    assert result.stdout == b"No new upgrade operations detected.\n"

    result = run_command("default_async.py add-user")
    assert result.returncode == 0
    assert result.stdout == b"<User name=default App Ellar id=1>\n"