Test alembic automigration with add and drop table and column. Args: tmp_working_dir: directory where database and migrations are stored monkeypatch: pytest fixture to overwrite attributes
(tmp_working_dir, monkeypatch)
| 64 | "ignore:This declarative base already contains a class with the same class name", |
| 65 | ) |
| 66 | def test_automigration(tmp_working_dir, monkeypatch): |
| 67 | """Test alembic automigration with add and drop table and column. |
| 68 | |
| 69 | Args: |
| 70 | tmp_working_dir: directory where database and migrations are stored |
| 71 | monkeypatch: pytest fixture to overwrite attributes |
| 72 | """ |
| 73 | alembic_ini = tmp_working_dir / "alembic.ini" |
| 74 | versions = tmp_working_dir / "alembic" / "versions" |
| 75 | monkeypatch.setattr(nextpy.constants, "ALEMBIC_CONFIG", str(alembic_ini)) |
| 76 | |
| 77 | config_mock = mock.Mock() |
| 78 | config_mock.db_url = f"sqlite:///{tmp_working_dir}/nextpy.db" |
| 79 | monkeypatch.setattr( |
| 80 | nextpy.data.model, "get_config", mock.Mock(return_value=config_mock) |
| 81 | ) |
| 82 | |
| 83 | Model.alembic_init() |
| 84 | assert alembic_ini.exists() |
| 85 | assert versions.exists() |
| 86 | |
| 87 | # initial table |
| 88 | class AlembicThing(Model, table=True): # type: ignore |
| 89 | t1: str |
| 90 | |
| 91 | with Model.get_db_engine().connect() as connection: |
| 92 | Model.alembic_autogenerate(connection=connection, message="Initial Revision") |
| 93 | Model.migrate() |
| 94 | version_scripts = list(versions.glob("*.py")) |
| 95 | assert len(version_scripts) == 1 |
| 96 | assert version_scripts[0].name.endswith("initial_revision.py") |
| 97 | |
| 98 | with nextpy.data.model.session() as session: |
| 99 | session.add(AlembicThing(id=None, t1="foo")) |
| 100 | session.commit() |
| 101 | |
| 102 | sqlmodel.SQLModel.metadata.clear() |
| 103 | |
| 104 | # Create column t2 |
| 105 | class AlembicThing(Model, table=True): # type: ignore |
| 106 | t1: str |
| 107 | t2: str = "bar" |
| 108 | |
| 109 | Model.migrate(autogenerate=True) |
| 110 | assert len(list(versions.glob("*.py"))) == 2 |
| 111 | |
| 112 | with nextpy.data.model.session() as session: |
| 113 | result = session.exec(sqlmodel.select(AlembicThing)).all() |
| 114 | assert len(result) == 1 |
| 115 | assert result[0].t2 == "bar" |
| 116 | |
| 117 | sqlmodel.SQLModel.metadata.clear() |
| 118 | |
| 119 | # Drop column t1 |
| 120 | class AlembicThing(Model, table=True): # type: ignore |
| 121 | t2: str = "bar" |
| 122 | |
| 123 | Model.migrate(autogenerate=True) |
nothing calls this directly
no test coverage detected