diff --git a/backend/app/models/alert_event.py b/backend/app/models/alert_event.py index e5587c2..4aa74c0 100644 --- a/backend/app/models/alert_event.py +++ b/backend/app/models/alert_event.py @@ -36,6 +36,11 @@ class AlertEvent(Base): nullable=True, comment="最近一次通知失败原因", ) + related_event_id = Column( + BigInteger, + nullable=True, + comment="关联的原始或升级告警事件 ID", + ) previous_status = Column(String(16), nullable=True, comment="状态变更前状态") current_status = Column(String(16), nullable=True, comment="状态变更后状态") acknowledged_at = Column(DateTime, nullable=True, comment="用户确认时间") diff --git a/backend/app/models/device.py b/backend/app/models/device.py index 19e1c0c..0e2b02b 100644 --- a/backend/app/models/device.py +++ b/backend/app/models/device.py @@ -36,6 +36,23 @@ class DeviceMonitoringPolicy: degraded_loss_percent: float recovery_consecutive_clean_rounds: int + @staticmethod + def _resolve_override( + override: object, + default: int | float, + minimum: int | float, + maximum: int | float, + ) -> int | float: + """Use a bounded persisted override or retain the validated default.""" + if ( + isinstance(override, bool) + or not isinstance(override, (int, float)) + or override < minimum + or override > maximum + ): + return default + return override + @classmethod def from_device( cls, @@ -45,29 +62,44 @@ class DeviceMonitoringPolicy: """Resolve device overrides without changing the stored device.""" return cls( probe_packets_per_round=( - device.probe_packets_per_round - if device.probe_packets_per_round is not None - else settings.probe_packets_per_round + cls._resolve_override( + device.probe_packets_per_round, + settings.probe_packets_per_round, + 1, + 10, + ) ), offline_consecutive_rounds=( - device.offline_consecutive_rounds - if device.offline_consecutive_rounds is not None - else settings.offline_consecutive_rounds + cls._resolve_override( + device.offline_consecutive_rounds, + settings.offline_consecutive_rounds, + 1, + 10, + ) ), degraded_window_rounds=( - device.degraded_window_rounds - if device.degraded_window_rounds is not None - else settings.degraded_window_rounds + cls._resolve_override( + device.degraded_window_rounds, + settings.degraded_window_rounds, + 2, + 60, + ) ), degraded_loss_percent=( - device.degraded_loss_percent - if device.degraded_loss_percent is not None - else settings.degraded_loss_percent + cls._resolve_override( + device.degraded_loss_percent, + settings.degraded_loss_percent, + 1, + 100, + ) ), recovery_consecutive_clean_rounds=( - device.recovery_consecutive_clean_rounds - if device.recovery_consecutive_clean_rounds is not None - else settings.recovery_consecutive_clean_rounds + cls._resolve_override( + device.recovery_consecutive_clean_rounds, + settings.recovery_consecutive_clean_rounds, + 1, + 20, + ) ), ) diff --git a/backend/migrations/versions/20260803_reliability_monitoring.py b/backend/migrations/versions/20260803_reliability_monitoring.py index 09c78d3..21b3048 100644 --- a/backend/migrations/versions/20260803_reliability_monitoring.py +++ b/backend/migrations/versions/20260803_reliability_monitoring.py @@ -5,7 +5,7 @@ from datetime import datetime from sqlalchemy import Column, DateTime, MetaData, String, Table, inspect, select, text from sqlalchemy.engine import Connection -from app.models import Base +from app.models import AlertEvent, Device, NotificationOutbox, PingRecord REVISION = "20260803_reliability_monitoring" @@ -23,6 +23,40 @@ _DEFAULTS_FOR_EXISTING_ROWS = { "notification_attempts": "0", } +_REVISION_COLUMNS = { + "devices": ( + Device.__table__, + { + "probe_packets_per_round", + "offline_consecutive_rounds", + "degraded_window_rounds", + "degraded_loss_percent", + "recovery_consecutive_clean_rounds", + }, + ), + "ping_records": ( + PingRecord.__table__, + { + "sent_count", + "received_count", + "packet_loss_percent", + "average_rtt_ms", + "is_valid", + "failure_reason", + }, + ), + "alert_events": ( + AlertEvent.__table__, + { + "notification_attempts", + "last_notification_error", + "related_event_id", + "previous_status", + "current_status", + }, + ), +} + def _quote(connection: Connection, identifier: str) -> str: """Quote one database identifier through the active dialect.""" @@ -33,15 +67,15 @@ def _add_missing_columns(connection: Connection) -> None: """Add only newly required nullable/defaulted columns to existing tables.""" inspector = inspect(connection) existing_tables = set(inspector.get_table_names()) - for table in Base.metadata.sorted_tables: - if table.name not in existing_tables: + for table_name, (table, column_names) in _REVISION_COLUMNS.items(): + if table_name not in existing_tables: continue existing_columns = { - column["name"] for column in inspector.get_columns(table.name) + column["name"] for column in inspector.get_columns(table_name) } for column in table.columns: - if column.name in existing_columns: + if column.name not in column_names or column.name in existing_columns: continue column_type = connection.dialect.type_compiler.process(column.type) @@ -50,10 +84,13 @@ def _add_missing_columns(connection: Connection) -> None: if default is not None: if connection.dialect.name == "sqlite" and default == "TRUE": default = "1" - definition = f"{definition} DEFAULT {default}" + if column.name == "is_valid": + definition = f"{definition} NOT NULL DEFAULT {default}" + else: + definition = f"{definition} DEFAULT {default}" connection.execute( text( - f"ALTER TABLE {_quote(connection, table.name)} " + f"ALTER TABLE {_quote(connection, table_name)} " f"ADD COLUMN {definition}" ) ) @@ -62,32 +99,40 @@ def _add_missing_columns(connection: Connection) -> None: def _create_missing_indexes(connection: Connection) -> None: """Create new performance indexes for tables that predate this revision.""" inspector = inspect(connection) - for table in Base.metadata.sorted_tables: - existing_indexes = { - index["name"] for index in inspector.get_indexes(table.name) - } - for index in table.indexes: - if index.name not in existing_indexes: - index.create(bind=connection) + table = PingRecord.__table__ + if table.name not in inspector.get_table_names(): + return + existing_indexes = { + index["name"] for index in inspector.get_indexes(table.name) + } + for index in table.indexes: + if index.name not in existing_indexes: + index.create(bind=connection) def _upgrade_postgresql_alert_type(connection: Connection) -> None: """Permit the degraded alert value for databases using PostgreSQL enums.""" if connection.dialect.name == "postgresql": + enum_name = AlertEvent.__table__.c.alert_type.type.name connection.execute( - text("ALTER TYPE alerttypeenum ADD VALUE IF NOT EXISTS 'degraded'") + text( + f"ALTER TYPE {_quote(connection, enum_name)} " + "ADD VALUE IF NOT EXISTS 'degraded'" + ) ) def run_reliability_migration(connection: Connection) -> None: """Apply this revision once while retaining all existing monitoring data. - This runner is deliberately idempotent: it creates only absent tables, - columns, and indexes, then writes one immutable revision marker. It never - issues DROP, DELETE, UPDATE, or data-copy statements. + This runner is deliberately idempotent: it creates only this revision's + outbox table, missing revision columns, and indexes, then writes one + immutable revision marker. It never issues DROP, DELETE, UPDATE, or + data-copy statements. The application owns global metadata creation after + this function returns. """ _migration_metadata.create_all(bind=connection) - Base.metadata.create_all(bind=connection) + NotificationOutbox.__table__.create(bind=connection, checkfirst=True) already_applied = connection.execute( select(_schema_migrations.c.revision).where( diff --git a/backend/tests/test_models_and_migration.py b/backend/tests/test_models_and_migration.py index 03934a9..fb6d7fa 100644 --- a/backend/tests/test_models_and_migration.py +++ b/backend/tests/test_models_and_migration.py @@ -4,7 +4,7 @@ import importlib from pathlib import Path import pytest -from sqlalchemy import text +from sqlalchemy import Column, Integer, Table, inspect, text from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine import app.models as models @@ -72,6 +72,19 @@ def test_device_policy_prefers_explicit_device_overrides(): assert policy.degraded_loss_percent == 20.0 +def test_device_policy_uses_global_default_for_invalid_persisted_override(): + """An invalid stored override cannot weaken a device's safe probe policy.""" + device = models.Device( + probe_packets_per_round=0, + degraded_loss_percent=101.0, + ) + + policy = models.DeviceMonitoringPolicy.from_device(device, Settings()) + + assert policy.probe_packets_per_round == 3 + assert policy.degraded_loss_percent == 20.0 + + async def test_migration_is_idempotent_and_records_revision(tmp_path: Path): """Rerunning the migration preserves the schema and writes one revision marker.""" migration = importlib.import_module( @@ -96,3 +109,143 @@ async def test_migration_is_idempotent_and_records_revision(tmp_path: Path): assert revision_count == 1 finally: await engine.dispose() + + +async def test_migration_preserves_legacy_rows_and_owns_only_its_schema( + tmp_path: Path, +): + """The revision upgrades legacy data twice without creating unrelated models.""" + migration = importlib.import_module( + "migrations.versions.20260803_reliability_monitoring" + ) + database_url = f"sqlite+aiosqlite:///{tmp_path / 'legacy-monitoring.db'}" + engine = create_async_engine(database_url) + future_table = Table( + "future_unrelated_model", + Base.metadata, + Column("id", Integer, primary_key=True), + ) + try: + async with engine.begin() as connection: + await connection.execute( + text( + "CREATE TABLE devices (" + "id INTEGER PRIMARY KEY, name VARCHAR(128) NOT NULL, " + "ip VARCHAR(45) NOT NULL, device_type VARCHAR(16) NOT NULL, " + "location VARCHAR(256), project_name VARCHAR(256), " + "tags VARCHAR(512), ping_interval INTEGER, " + "alert_threshold INTEGER, is_enabled BOOLEAN, " + "current_status VARCHAR(16), consecutive_failures INTEGER, " + "last_ping_time DATETIME, last_online_time DATETIME, " + "last_offline_time DATETIME, created_at DATETIME, " + "updated_at DATETIME)" + ) + ) + await connection.execute( + text( + "CREATE TABLE ping_records (" + "id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, " + "is_alive BOOLEAN NOT NULL, response_time_ms FLOAT, " + "round_num INTEGER NOT NULL, created_at DATETIME)" + ) + ) + await connection.execute( + text( + "CREATE TABLE alert_events (" + "id INTEGER PRIMARY KEY, device_id INTEGER NOT NULL, " + "alert_type VARCHAR(16) NOT NULL, message VARCHAR(1024), " + "start_at DATETIME NOT NULL, end_at DATETIME, " + "duration_minutes INTEGER, is_resolved BOOLEAN, " + "notification_sent BOOLEAN, acknowledged_at DATETIME, " + "created_at DATETIME)" + ) + ) + await connection.execute( + text( + "INSERT INTO devices (id, name, ip, device_type, current_status) " + "VALUES (1, 'legacy-router', '192.0.2.10', 'router', 'online')" + ) + ) + await connection.execute( + text( + "INSERT INTO ping_records " + "(id, device_id, is_alive, response_time_ms, round_num) " + "VALUES (2, 1, 1, 8.5, 7)" + ) + ) + await connection.execute( + text( + "INSERT INTO alert_events " + "(id, device_id, alert_type, message, start_at, notification_sent) " + "VALUES (3, 1, 'offline', 'legacy event', CURRENT_TIMESTAMP, 1)" + ) + ) + + await connection.run_sync(migration.run_reliability_migration) + table_names = await connection.run_sync( + lambda sync_connection: set( + inspect(sync_connection).get_table_names() + ) + ) + assert "future_unrelated_model" not in table_names + assert "notification_outbox" in table_names + Base.metadata.remove(future_table) + await connection.run_sync(Base.metadata.create_all) + + async with engine.begin() as connection: + await connection.run_sync(migration.run_reliability_migration) + legacy_device = await connection.execute( + text("SELECT name, ip, current_status FROM devices WHERE id = 1") + ) + legacy_probe = await connection.execute( + text( + "SELECT device_id, is_alive, response_time_ms, round_num " + "FROM ping_records WHERE id = 2" + ) + ) + legacy_alert = await connection.execute( + text( + "SELECT device_id, alert_type, message, notification_sent " + "FROM alert_events WHERE id = 3" + ) + ) + probe_columns, alert_columns, probe_indexes = await connection.run_sync( + lambda sync_connection: ( + { + column["name"]: column + for column in inspect(sync_connection).get_columns( + "ping_records" + ) + }, + { + column["name"] + for column in inspect(sync_connection).get_columns( + "alert_events" + ) + }, + { + index["name"] + for index in inspect(sync_connection).get_indexes( + "ping_records" + ) + }, + ) + ) + + assert legacy_device.one() == ("legacy-router", "192.0.2.10", "online") + assert legacy_probe.one() == (1, 1, 8.5, 7) + assert legacy_alert.one() == (1, "offline", "legacy event", 1) + assert probe_columns["is_valid"]["nullable"] is False + assert { + "sent_count", + "received_count", + "packet_loss_percent", + "average_rtt_ms", + "failure_reason", + }.issubset(probe_columns) + assert "related_event_id" in alert_columns + assert "idx_ping_records_device_created_at" in probe_indexes + finally: + if future_table in Base.metadata.tables.values(): + Base.metadata.remove(future_table) + await engine.dispose()