|
1 | 1 | """
|
2 | 2 | Test Alembic migrations
|
3 | 3 | """
|
4 |
| -import subprocess |
| 4 | +import os |
| 5 | +from contextlib import contextmanager |
| 6 | +from os import fspath |
| 7 | +from pathlib import Path |
| 8 | +from typing import Iterator |
5 | 9 |
|
6 |
| -from passari_workflow.db.models import Base |
7 |
| -from alembic.config import Config |
8 | 10 | from alembic import command
|
| 11 | +from alembic.config import Config |
9 | 12 |
|
10 |
| -from pathlib import Path |
| 13 | +from passari_workflow.db.models import Base |
| 14 | + |
| 15 | +ROOT_PATH = Path(__file__).resolve().parent.parent.parent |
11 | 16 |
|
12 | 17 |
|
13 | 18 | def test_migrate(session, engine, database):
|
14 | 19 | """
|
15 | 20 | Test migrations by running all migrations and then downgrading
|
16 | 21 | """
|
17 |
| - config = Config( |
18 |
| - str(Path(__file__).resolve().parent.parent.parent / "alembic.ini") |
19 |
| - ) |
| 22 | + config = Config(fspath(ROOT_PATH / "alembic.ini")) |
20 | 23 |
|
21 | 24 | try:
|
22 | 25 | Base.metadata.drop_all(engine)
|
23 | 26 |
|
24 |
| - # Upgrade the database |
25 |
| - command.upgrade(config, "head") |
| 27 | + with run_in_path(ROOT_PATH): |
| 28 | + # Upgrade the database |
| 29 | + command.upgrade(config, "head") |
26 | 30 |
|
27 |
| - # Downgrade the database |
28 |
| - command.downgrade(config, "base") |
| 31 | + # Downgrade the database |
| 32 | + command.downgrade(config, "base") |
29 | 33 | finally:
|
30 | 34 | # Ensure that the tables are the same after the test, even if
|
31 | 35 | # something fails
|
32 | 36 | Base.metadata.drop_all(engine)
|
33 | 37 | Base.metadata.create_all(engine)
|
| 38 | + |
| 39 | + |
| 40 | +@contextmanager |
| 41 | +def run_in_path(path: Path) -> Iterator[None]: |
| 42 | + orig_cwd = os.getcwd() |
| 43 | + try: |
| 44 | + os.chdir(path) |
| 45 | + yield |
| 46 | + finally: |
| 47 | + os.chdir(orig_cwd) |
0 commit comments