130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
"""
|
|
После Alembic upgrade (ревизия wesp_2_0) SQLite содержит все колонки ORM: recipes и reports.
|
|
|
|
Проверка живой развёртки (без upgrade):
|
|
|
|
WESP_SCHEMA_CHECK_LIVE_DB=1 \\
|
|
WESP_RECIPES_DB_URI=sqlite:////path/to/recipes.db \\
|
|
WESP_REPORTS_DB_URI=sqlite:////path/to/reports.db \\
|
|
python3 -m pytest tests/test_migrated_schema_matches_models.py -q
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import inspect as sa_inspect
|
|
|
|
from app import create_app, db
|
|
from config import Config
|
|
|
|
|
|
def _iter_main_bind_models():
|
|
import app.models # noqa: F401 — регистрация мапперов
|
|
|
|
for mapper in db.Model.registry.mappers:
|
|
cls = mapper.class_
|
|
if getattr(cls, "__bind_key__", None) == "reports":
|
|
continue
|
|
if cls.__dict__.get("__abstract__", False):
|
|
continue
|
|
tab = getattr(cls, "__tablename__", None)
|
|
if not tab:
|
|
continue
|
|
yield cls
|
|
|
|
|
|
def _iter_reports_bind_models():
|
|
import app.models # noqa: F401
|
|
|
|
for mapper in db.Model.registry.mappers:
|
|
cls = mapper.class_
|
|
if getattr(cls, "__bind_key__", None) != "reports":
|
|
continue
|
|
if cls.__dict__.get("__abstract__", False):
|
|
continue
|
|
tab = getattr(cls, "__tablename__", None)
|
|
if not tab:
|
|
continue
|
|
yield cls
|
|
|
|
|
|
def _assert_columns_match(engine, model_iter, label: str) -> None:
|
|
insp = sa_inspect(engine)
|
|
missing: list[str] = []
|
|
for cls in model_iter:
|
|
tab = cls.__tablename__
|
|
if not insp.has_table(tab):
|
|
missing.append(f"{cls.__name__}: table {tab!r} missing")
|
|
continue
|
|
actual = {c["name"] for c in insp.get_columns(tab)}
|
|
for col in cls.__table__.columns:
|
|
if col.key not in actual:
|
|
missing.append(f"{cls.__name__}.{col.key}: column missing in {tab}")
|
|
assert not missing, f"{label}:\n" + "\n".join(missing)
|
|
|
|
|
|
class MigratedSchemaMatchesModelsTest(unittest.TestCase):
|
|
"""Свежие файлы SQLite + полный upgrade head."""
|
|
|
|
def setUp(self) -> None:
|
|
self._tmp = tempfile.mkdtemp(prefix="wesp-schema-migration-")
|
|
recipes = Path(self._tmp) / "recipes.db"
|
|
reports = Path(self._tmp) / "reports.db"
|
|
|
|
class C(Config):
|
|
TESTING = False
|
|
WESP_BACKGROUND_STARTUP = False
|
|
SQLALCHEMY_DATABASE_URI = f"sqlite:///{recipes}"
|
|
SQLALCHEMY_BINDS = {"reports": f"sqlite:///{reports}"}
|
|
SECRET_KEY = "schema-test-secret"
|
|
SYNC_BACKGROUND_REQUEUE = False
|
|
SYNC_CLIENT_AUTOSTART = False
|
|
WESP_ADMIN_LOG_PATH = ""
|
|
WESP_ADMIN_HARDWARE_LOG_PATH = ""
|
|
WESP_ADMIN_UI_ACTIVITY_LOG_PATH = ""
|
|
|
|
self.app = create_app(C, run_migrations=True)
|
|
self._ctx = self.app.app_context()
|
|
self._ctx.push()
|
|
|
|
def tearDown(self) -> None:
|
|
self._ctx.pop()
|
|
|
|
def test_recipes_db_columns_cover_all_main_models(self) -> None:
|
|
_assert_columns_match(db.engine, _iter_main_bind_models(), "recipes.db / main bind")
|
|
|
|
def test_reports_db_columns_cover_reports_models(self) -> None:
|
|
engine = db.engines["reports"]
|
|
_assert_columns_match(engine, _iter_reports_bind_models(), "reports.db")
|
|
|
|
|
|
@unittest.skipUnless(
|
|
os.getenv("WESP_SCHEMA_CHECK_LIVE_DB", "").strip() in {"1", "true", "yes"},
|
|
"Задайте WESP_SCHEMA_CHECK_LIVE_DB=1 и WESP_RECIPES_DB_URI для проверки живой БД",
|
|
)
|
|
class LiveRecipesDbSchemaTest(unittest.TestCase):
|
|
"""Опционально: проверка существующих файлов БД (без upgrade)."""
|
|
|
|
def test_live_db_columns_cover_models(self) -> None:
|
|
uri = (os.getenv("WESP_RECIPES_DB_URI") or "").strip()
|
|
self.assertTrue(uri, "Нужен WESP_RECIPES_DB_URI для живой проверки")
|
|
reports_uri = (os.getenv("WESP_REPORTS_DB_URI") or "").strip()
|
|
|
|
class C(Config):
|
|
SQLALCHEMY_DATABASE_URI = uri
|
|
SQLALCHEMY_BINDS = {"reports": reports_uri} if reports_uri else {}
|
|
TESTING = True
|
|
SYNC_BACKGROUND_REQUEUE = False
|
|
WESP_ADMIN_LOG_PATH = ""
|
|
|
|
app = create_app(C, run_migrations=False)
|
|
with app.app_context():
|
|
_assert_columns_match(db.engine, _iter_main_bind_models(), "живая recipes.db")
|
|
if reports_uri:
|
|
eng = db.engines["reports"]
|
|
_assert_columns_match(eng, _iter_reports_bind_models(), "живая reports.db")
|