From d9c82e39fb5b7d41d75613de14c0645a0f864f23 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Mon, 22 Jun 2026 12:45:50 +0200 Subject: [PATCH] M5-T02: add modbus_device and modbus_reading tables and models --- alembic_app/env.py | 1 + .../versions/20260622_09_modbus_tables.py | 81 ++++ app/models/modbus.py | 114 +++++ docs/design/m5-iot-energy.md | 2 +- scripts/app_db_adopt.py | 2 +- tests/test_modbus_models.py | 396 ++++++++++++++++++ tests/test_totp_models.py | 6 +- 7 files changed, 598 insertions(+), 4 deletions(-) create mode 100644 alembic_app/versions/20260622_09_modbus_tables.py create mode 100644 app/models/modbus.py create mode 100644 tests/test_modbus_models.py diff --git a/alembic_app/env.py b/alembic_app/env.py index c23bde5..56c4383 100644 --- a/alembic_app/env.py +++ b/alembic_app/env.py @@ -11,6 +11,7 @@ from app.models.auth_throttle import LoginThrottle # noqa: F401 from app.models.public_ip import PublicIPHistory, PublicIPState # noqa: F401 from app.models.location import Location # noqa: F401 from app.models.poo import PooRecord # noqa: F401 +from app.models.modbus import ModbusDevice, ModbusReading # noqa: F401 config = context.config diff --git a/alembic_app/versions/20260622_09_modbus_tables.py b/alembic_app/versions/20260622_09_modbus_tables.py new file mode 100644 index 0000000..b531e3d --- /dev/null +++ b/alembic_app/versions/20260622_09_modbus_tables.py @@ -0,0 +1,81 @@ +"""add modbus_device and modbus_reading tables + +Revision ID: 20260622_09_modbus_tables +Revises: 20260621_08_totp +Create Date: 2026-06-22 00:00:00.000000 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +revision: str = "20260622_09_modbus_tables" +down_revision: Union[str, None] = "20260621_08_totp" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # modbus_device — deployment/configurable metadata for each polled device. + op.create_table( + "modbus_device", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("uuid", sa.String(length=36), nullable=False), + sa.Column("friendly_name", sa.String(length=255), nullable=False), + sa.Column("transport", sa.String(length=16), nullable=False), + sa.Column("host", sa.String(length=255), nullable=False), + sa.Column("port", sa.Integer(), nullable=False), + sa.Column("unit_id", sa.Integer(), nullable=False), + sa.Column("profile", sa.String(length=64), nullable=False), + sa.Column("poll_interval_s", sa.Integer(), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column("last_poll_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_poll_ok", sa.Boolean(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("uuid", name="uq_modbus_device_uuid"), + ) + + # modbus_reading — generic telemetry, one row per device per poll cycle. + op.create_table( + "modbus_reading", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("device_id", sa.Integer(), nullable=False), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("payload", sa.JSON(), nullable=False), + sa.ForeignKeyConstraint( + ["device_id"], + ["modbus_device.id"], + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id"), + ) + + # Individual index on recorded_at (from the ORM-level index=True). + op.create_index( + "ix_modbus_reading_recorded_at", + "modbus_reading", + ["recorded_at"], + unique=False, + ) + + # Composite index for efficient time-range queries per device. + op.create_index( + "ix_modbus_reading_device_recorded", + "modbus_reading", + ["device_id", "recorded_at"], + unique=False, + ) + + +def downgrade() -> None: + # Drop the reading table first (it has a FK referencing modbus_device). + op.drop_index("ix_modbus_reading_device_recorded", table_name="modbus_reading") + op.drop_index("ix_modbus_reading_recorded_at", table_name="modbus_reading") + op.drop_table("modbus_reading") + + # Drop the device table. + op.drop_table("modbus_device") diff --git a/app/models/modbus.py b/app/models/modbus.py new file mode 100644 index 0000000..acd0533 --- /dev/null +++ b/app/models/modbus.py @@ -0,0 +1,114 @@ +"""SQLAlchemy models for Modbus device management and telemetry. + +Two tables: + - modbus_device: deployment/configurable metadata for each polled device. + - modbus_reading: generic telemetry rows (one per device per poll cycle). +""" + +from __future__ import annotations + +import uuid as _uuid +from datetime import datetime + +from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, String +from sqlalchemy.orm import Mapped, mapped_column, relationship +from sqlalchemy.types import JSON + +from app.db import Base + + +def _uuid4_str() -> str: + return str(_uuid.uuid4()) + + +class ModbusDevice(Base): + """Deployment-layer record for a Modbus slave device reachable via a TCP gateway. + + Protocol knowledge (register map, decoding rules) lives in the YAML profile + referenced by ``profile``. Per-device variables (network address, slave ID, + display name, poll rate) live here. + """ + + __tablename__ = "modbus_device" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # Stable internal identity — used as the API path key and HA Discovery unique_id anchor. + uuid: Mapped[str] = mapped_column( + String(36), unique=True, nullable=False, default=_uuid4_str + ) + + # Human-readable label (may be changed; changing it triggers re-publish of HA Discovery). + friendly_name: Mapped[str] = mapped_column(String(255), nullable=False) + + # Transport protocol — only "tcp" is supported for now. + transport: Mapped[str] = mapped_column(String(16), nullable=False, default="tcp") + + # TCP gateway address. + host: Mapped[str] = mapped_column(String(255), nullable=False) + port: Mapped[int] = mapped_column(Integer, nullable=False, default=502) + + # Modbus slave address (set on the device panel; different meters on the same gateway + # must have distinct unit_ids). + unit_id: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + + # Which YAML profile to use for register mapping and decoding (e.g. "sdm120"). + profile: Mapped[str] = mapped_column(String(64), nullable=False) + + # Polling interval in seconds. + poll_interval_s: Mapped[int] = mapped_column(Integer, nullable=False, default=5) + + # Whether this device is included in the periodic poll sweep. + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + + # Poll-status fields (updated by the poll service after each attempt). + last_poll_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + last_poll_ok: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + + # Audit timestamps. + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False + ) + + # Relationship to readings (back-reference; not loaded eagerly). + readings: Mapped[list["ModbusReading"]] = relationship( + back_populates="device", cascade="save-update, merge" + ) + + +class ModbusReading(Base): + """Generic telemetry row: one device, one poll instant, all decoded metrics as JSON. + + ``payload`` contains the full dict of engineering values produced by the YAML profile + decoder, e.g. ``{"voltage": 230.2, "current": 1.3, ...}``. The profile is the key + to interpreting which keys exist and what units they carry. + """ + + __tablename__ = "modbus_reading" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # FK to the device that produced this reading. + # ON DELETE RESTRICT: prevents accidental deletion of a device that has historical data. + device_id: Mapped[int] = mapped_column( + ForeignKey("modbus_device.id", ondelete="RESTRICT"), nullable=False + ) + + # The UTC timestamp of when the sample was taken — real indexed column, NOT embedded in payload. + recorded_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, index=True + ) + + # Profile-decoded engineering values as a JSON object. + payload: Mapped[dict] = mapped_column(JSON, nullable=False) + + device: Mapped["ModbusDevice"] = relationship(back_populates="readings") + + +# Composite index for efficient time-range queries scoped to a single device. +Index("ix_modbus_reading_device_recorded", ModbusReading.device_id, ModbusReading.recorded_at) diff --git a/docs/design/m5-iot-energy.md b/docs/design/m5-iot-energy.md index b38aff0..9077565 100644 --- a/docs/design/m5-iot-energy.md +++ b/docs/design/m5-iot-energy.md @@ -290,7 +290,7 @@ Phase C（MQTT / Discovery，依赖 B 的设备数据与 provider 接口） - **Reviewer checklist**: 仅改 ConfigPage（+其测试），未碰 config 后端或主 layout；保存逻辑无回归；页面内**无第二个 app 级 sidebar**（accordion 是内容、非 chrome）。 ### M5-T02 — `modbus_device` + `modbus_reading` 表与模型 `[schema]` -- **Status**: `todo` · **Depends**: none +- **Status**: `done` · **Depends**: none - **Context**: 单库 app 链新增两张**通用**表，建出 §3.2 结构（设备 = 部署层，读数 = JSON payload 通用遥测层）。本任务只建 schema + 模型，不写采集/接口。 - **Files**: `create app/models/modbus.py`（`ModbusDevice`、`ModbusReading`，继承 `app.db.Base`）；`create alembic_app/versions/_07_modbus_tables.py`；`modify alembic_app/env.py`（import 新模型）；`modify scripts/app_db_adopt.py`（`APP_BASELINE_REVISION` → 新 head）；`create tests/test_modbus_models.py` - **Steps**: diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index 257b021..1ca952f 100644 --- a/scripts/app_db_adopt.py +++ b/scripts/app_db_adopt.py @@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: from app.config import get_settings -APP_BASELINE_REVISION = "20260621_08_totp" +APP_BASELINE_REVISION = "20260622_09_modbus_tables" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_modbus_models.py b/tests/test_modbus_models.py new file mode 100644 index 0000000..0231df9 --- /dev/null +++ b/tests/test_modbus_models.py @@ -0,0 +1,396 @@ +"""Tests for M5-T02: modbus_device + modbus_reading tables and ORM models. + +Covers: + 1. Migration shape: upgrade to head creates both tables with correct columns, + constraints, and indexes; downgrade -1 cleanly removes them. + 2. ORM metadata: Base.metadata.tables contains both tables; FK is RESTRICT; + uuid column is unique and auto-generated. + 3. Baseline constant: APP_BASELINE_REVISION matches the actual Alembic head. + 4. Basic ORM round-trip: insert + retrieve, uuid auto-population, nullable fields. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, inspect, text +from sqlalchemy.orm import Session + +from app.db import Base +from app.models.modbus import ModbusDevice, ModbusReading +from scripts.app_db_adopt import APP_BASELINE_REVISION + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_app_alembic_config(database_url: str) -> Config: + cfg = Config("alembic_app.ini") + cfg.set_main_option("sqlalchemy.url", database_url) + return cfg + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def modbus_db(tmp_path: Path): + """Temporary SQLite DB upgraded to the current Alembic head.""" + db_path = tmp_path / "modbus_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + yield engine + engine.dispose() + + +# --------------------------------------------------------------------------- +# 1. Migration shape tests +# --------------------------------------------------------------------------- + + +def test_modbus_tables_exist_after_upgrade(modbus_db): + """Both modbus_device and modbus_reading tables must exist after upgrade to head.""" + inspector = inspect(modbus_db) + table_names = inspector.get_table_names() + assert "modbus_device" in table_names, "modbus_device table missing after upgrade" + assert "modbus_reading" in table_names, "modbus_reading table missing after upgrade" + + +def test_modbus_device_columns(modbus_db): + """modbus_device must have all required columns with correct nullability.""" + inspector = inspect(modbus_db) + columns = {col["name"]: col for col in inspector.get_columns("modbus_device")} + + expected_non_nullable = { + "id", "uuid", "friendly_name", "transport", "host", "port", + "unit_id", "profile", "poll_interval_s", "enabled", + "created_at", "updated_at", + } + expected_nullable = {"last_poll_at", "last_poll_ok"} + + for col_name in expected_non_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert not columns[col_name]["nullable"], f"Column {col_name} should be NOT NULL" + + for col_name in expected_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert columns[col_name]["nullable"], f"Column {col_name} should be nullable" + + +def test_modbus_reading_columns(modbus_db): + """modbus_reading must have id, device_id, recorded_at, payload with correct nullability.""" + inspector = inspect(modbus_db) + columns = {col["name"]: col for col in inspector.get_columns("modbus_reading")} + + expected_non_nullable = {"id", "device_id", "recorded_at", "payload"} + for col_name in expected_non_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert not columns[col_name]["nullable"], f"Column {col_name} should be NOT NULL" + + +def test_modbus_device_uuid_is_unique(modbus_db): + """modbus_device.uuid must have a unique constraint.""" + inspector = inspect(modbus_db) + unique_constraints = inspector.get_unique_constraints("modbus_device") + unique_cols = [ + col + for uc in unique_constraints + for col in uc["column_names"] + ] + assert "uuid" in unique_cols, "uuid column must have a unique constraint" + + +def test_modbus_reading_fk_to_device(modbus_db): + """modbus_reading.device_id must have a FK referencing modbus_device.id.""" + inspector = inspect(modbus_db) + fks = inspector.get_foreign_keys("modbus_reading") + assert len(fks) == 1, f"Expected 1 FK on modbus_reading, got {len(fks)}" + fk = fks[0] + assert fk["referred_table"] == "modbus_device" + assert "device_id" in fk["constrained_columns"] + assert "id" in fk["referred_columns"] + + +def test_modbus_reading_composite_index_exists(modbus_db): + """Composite index (device_id, recorded_at) must exist on modbus_reading.""" + inspector = inspect(modbus_db) + indexes = {idx["name"]: idx for idx in inspector.get_indexes("modbus_reading")} + assert "ix_modbus_reading_device_recorded" in indexes, ( + "Composite index ix_modbus_reading_device_recorded missing" + ) + composite_idx = indexes["ix_modbus_reading_device_recorded"] + assert "device_id" in composite_idx["column_names"] + assert "recorded_at" in composite_idx["column_names"] + + +def test_modbus_reading_recorded_at_index_exists(modbus_db): + """Individual index on recorded_at must exist on modbus_reading.""" + inspector = inspect(modbus_db) + indexes = {idx["name"]: idx for idx in inspector.get_indexes("modbus_reading")} + # The ORM-level index=True creates ix_modbus_reading_recorded_at + assert "ix_modbus_reading_recorded_at" in indexes, ( + "Index ix_modbus_reading_recorded_at missing from modbus_reading" + ) + + +def test_downgrade_removes_modbus_tables(tmp_path: Path): + """downgrade -1 from head must drop both modbus tables cleanly.""" + db_path = tmp_path / "downgrade_modbus_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + + command.upgrade(alembic_cfg, "head") + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + inspector = inspect(engine) + assert "modbus_device" in inspector.get_table_names() + assert "modbus_reading" in inspector.get_table_names() + engine.dispose() + + # Downgrade one step removes the modbus migration. + command.downgrade(alembic_cfg, "-1") + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + inspector = inspect(engine) + table_names = inspector.get_table_names() + assert "modbus_device" not in table_names, "modbus_device should be gone after downgrade" + assert "modbus_reading" not in table_names, "modbus_reading should be gone after downgrade" + engine.dispose() + + +# --------------------------------------------------------------------------- +# 2. ORM metadata checks (Base.metadata) +# --------------------------------------------------------------------------- + + +def test_base_metadata_contains_modbus_tables(): + """Base.metadata.tables must include both new modbus tables.""" + assert "modbus_device" in Base.metadata.tables, "modbus_device not in Base.metadata" + assert "modbus_reading" in Base.metadata.tables, "modbus_reading not in Base.metadata" + + +def test_modbus_reading_fk_ondelete_restrict(): + """The FK from modbus_reading.device_id to modbus_device.id must be ON DELETE RESTRICT.""" + reading_table = Base.metadata.tables["modbus_reading"] + fk_columns = { + col.name: col + for col in reading_table.columns + } + device_id_col = fk_columns["device_id"] + assert device_id_col.foreign_keys, "device_id must have a foreign key" + fk = next(iter(device_id_col.foreign_keys)) + assert fk.ondelete == "RESTRICT", ( + f"FK ondelete must be RESTRICT, got: {fk.ondelete!r}" + ) + + +def test_modbus_device_uuid_unique_constraint_in_metadata(): + """ModbusDevice.uuid column must be declared unique in ORM metadata.""" + device_table = Base.metadata.tables["modbus_device"] + uuid_col = device_table.columns["uuid"] + assert uuid_col.unique, "ModbusDevice.uuid must be declared unique" + + +# --------------------------------------------------------------------------- +# 3. Baseline constant +# --------------------------------------------------------------------------- + + +def test_app_baseline_revision_matches_head(tmp_path: Path): + """APP_BASELINE_REVISION must equal the Alembic head revision.""" + from alembic.script import ScriptDirectory + + db_url = f"sqlite:///{tmp_path / 'rev_check.db'}" + alembic_cfg = _make_app_alembic_config(db_url) + script = ScriptDirectory.from_config(alembic_cfg) + heads = script.get_heads() + assert len(heads) == 1, f"Expected exactly 1 Alembic head, got {heads}" + head = heads[0] + assert APP_BASELINE_REVISION == head, ( + f"APP_BASELINE_REVISION={APP_BASELINE_REVISION!r} does not match " + f"Alembic head={head!r}" + ) + + +# --------------------------------------------------------------------------- +# 4. ORM round-trip tests +# --------------------------------------------------------------------------- + + +def test_modbus_device_uuid_auto_generated(modbus_db): + """uuid must be auto-populated when not explicitly provided.""" + now = datetime.now(tz=timezone.utc) + with Session(modbus_db) as session: + device = ModbusDevice( + friendly_name="Test Meter", + host="192.168.1.100", + profile="sdm120", + created_at=now, + updated_at=now, + ) + session.add(device) + session.commit() + device_id = device.id + + with Session(modbus_db) as session: + fetched = session.get(ModbusDevice, device_id) + assert fetched is not None + assert fetched.uuid is not None, "uuid should be auto-populated" + assert len(fetched.uuid) == 36, "uuid should be a standard UUID4 string (36 chars)" + # Verify it's a valid UUID4 format + import uuid as _uuid_mod + parsed = _uuid_mod.UUID(fetched.uuid) + assert parsed.version == 4, "uuid should be version 4" + + +def test_modbus_device_defaults(modbus_db): + """Default values for port, unit_id, poll_interval_s, enabled, transport must apply.""" + now = datetime.now(tz=timezone.utc) + with Session(modbus_db) as session: + device = ModbusDevice( + friendly_name="Default Test", + host="10.0.0.1", + profile="sdm120", + created_at=now, + updated_at=now, + ) + session.add(device) + session.commit() + device_id = device.id + + with Session(modbus_db) as session: + fetched = session.get(ModbusDevice, device_id) + assert fetched is not None + assert fetched.port == 502 + assert fetched.unit_id == 1 + assert fetched.poll_interval_s == 5 + assert fetched.enabled is True + assert fetched.transport == "tcp" + assert fetched.last_poll_at is None + assert fetched.last_poll_ok is None + + +def test_modbus_device_two_unique_uuids(modbus_db): + """Two independently created devices must have different auto-generated UUIDs.""" + now = datetime.now(tz=timezone.utc) + with Session(modbus_db) as session: + d1 = ModbusDevice( + friendly_name="Device 1", host="10.0.0.1", profile="sdm120", + created_at=now, updated_at=now, + ) + d2 = ModbusDevice( + friendly_name="Device 2", host="10.0.0.2", profile="sdm120", + created_at=now, updated_at=now, + ) + session.add_all([d1, d2]) + session.commit() + assert d1.uuid != d2.uuid, "Two devices must receive distinct UUIDs" + + +def test_modbus_reading_insert_and_retrieve(modbus_db): + """A ModbusReading can be inserted with a JSON payload and retrieved correctly.""" + now = datetime.now(tz=timezone.utc) + sample_payload = { + "voltage": 230.2, + "current": 1.3, + "active_power": 295.0, + "power_factor": 0.98, + "frequency": 50.0, + "import_energy": 123.4, + "export_energy": 0.0, + "total_energy": 123.4, + } + + with Session(modbus_db) as session: + device = ModbusDevice( + friendly_name="SDM120 AC", + host="192.168.1.100", + profile="sdm120", + created_at=now, + updated_at=now, + ) + session.add(device) + session.flush() + + reading = ModbusReading( + device_id=device.id, + recorded_at=now, + payload=sample_payload, + ) + session.add(reading) + session.commit() + reading_id = reading.id + + with Session(modbus_db) as session: + fetched = session.get(ModbusReading, reading_id) + assert fetched is not None + assert fetched.recorded_at is not None + assert fetched.payload == sample_payload + assert fetched.payload["voltage"] == 230.2 + assert fetched.payload["frequency"] == 50.0 + + +def test_modbus_reading_restrict_prevents_device_deletion(tmp_path: Path): + """Deleting a device with readings must fail due to ON DELETE RESTRICT. + + SQLite only enforces FK constraints when ``PRAGMA foreign_keys = ON`` is set + on the connection, so we build a dedicated engine that enables this pragma. + """ + import sqlalchemy.exc + from sqlalchemy import event as sa_event + + db_path = tmp_path / "restrict_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + + # Enable FK enforcement for every connection on this engine. + @sa_event.listens_for(engine, "connect") + def _enable_fk(dbapi_conn, _rec): + cursor = dbapi_conn.cursor() + cursor.execute("PRAGMA foreign_keys = ON") + cursor.close() + + now = datetime.now(tz=timezone.utc) + with Session(engine) as session: + device = ModbusDevice( + friendly_name="Restricted Device", + host="10.0.0.1", + profile="sdm120", + created_at=now, + updated_at=now, + ) + session.add(device) + session.flush() + + reading = ModbusReading( + device_id=device.id, + recorded_at=now, + payload={"voltage": 230.0}, + ) + session.add(reading) + session.commit() + device_id = device.id + + # Attempting to delete the device should raise an IntegrityError. + with pytest.raises(sqlalchemy.exc.IntegrityError): + with Session(engine) as session: + session.execute( + text("DELETE FROM modbus_device WHERE id = :did"), + {"did": device_id}, + ) + session.commit() + + engine.dispose() diff --git a/tests/test_totp_models.py b/tests/test_totp_models.py index 8805c9d..6ca9cd1 100644 --- a/tests/test_totp_models.py +++ b/tests/test_totp_models.py @@ -262,8 +262,10 @@ def test_downgrade_removes_totp_columns_and_table(tmp_path: Path): assert "totp_enabled" in auth_user_cols engine.dispose() - # Downgrade one step (removes TOTP migration). - command.downgrade(alembic_cfg, "-1") + # Downgrade to the revision before TOTP (removes TOTP migration). + # We use an explicit target rather than "-1" so the test stays correct even + # as new revisions are added on top of the TOTP revision. + command.downgrade(alembic_cfg, "20260621_07_auth_login_throttle") engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine)