From 49d15d8ffe1c2bfd32fdfa02b94e05545a846017 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Mon, 22 Jun 2026 13:24:11 +0200 Subject: [PATCH] M5-T04: add Modbus poll service and APScheduler polling job --- app/config.py | 3 + app/main.py | 18 ++ app/services/modbus_poll.py | 184 ++++++++++++++ docs/design/m5-iot-energy.md | 2 +- tests/test_modbus_poll.py | 465 +++++++++++++++++++++++++++++++++++ 5 files changed, 671 insertions(+), 1 deletion(-) create mode 100644 app/services/modbus_poll.py create mode 100644 tests/test_modbus_poll.py diff --git a/app/config.py b/app/config.py index 24c5b74..6228d4a 100644 --- a/app/config.py +++ b/app/config.py @@ -41,6 +41,9 @@ class Settings(BaseSettings): auth_trust_forwarded_for: bool = False auth_totp_issuer: str = "" # defaults to app_name when empty + # Modbus polling — global kill-switch; T08 will wire this into CONFIG_FIELDS/UI. + modbus_polling_enabled: bool = True + model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", diff --git a/app/main.py b/app/main.py index 267fb6e..2a4b91f 100644 --- a/app/main.py +++ b/app/main.py @@ -25,6 +25,7 @@ from app.config import get_settings from app.services.auth import AuthBootstrapError, initialize_auth_schema from app.services.config_page import seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap from app.services.public_ip import check_public_ipv4_and_notify +from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db logger = logging.getLogger(__name__) @@ -48,6 +49,15 @@ def _run_scheduled_public_ip_check() -> None: session.close() +def _run_scheduled_modbus_poll() -> None: + session_local = get_session_local() + session: Session = session_local() + try: + poll_all_enabled_devices(session, bootstrap_settings=get_settings()) + finally: + session.close() + + def ensure_auth_db_ready() -> None: session_local = get_session_local() session: Session = session_local() @@ -83,6 +93,14 @@ async def lifespan(_: FastAPI): max_instances=1, coalesce=True, ) + scheduler.add_job( + _run_scheduled_modbus_poll, + trigger=IntervalTrigger(seconds=BASE_POLL_TICK_SECONDS), + id="modbus-poll", + replace_existing=True, + max_instances=1, + coalesce=True, + ) scheduler.start() yield scheduler.shutdown(wait=False) diff --git a/app/services/modbus_poll.py b/app/services/modbus_poll.py new file mode 100644 index 0000000..9e39e7f --- /dev/null +++ b/app/services/modbus_poll.py @@ -0,0 +1,184 @@ +"""Modbus polling service — periodic read-and-store for enabled Modbus devices. + +Design notes +------------ +- ``poll_device`` loads the device's YAML profile, calls the driver to bulk-read + all register blocks in a single TCP connection, decodes the raw registers into + an engineering-value dict via the profile, and persists one ``ModbusReading`` + row. It also stamps ``last_poll_at`` / ``last_poll_ok`` on the device. +- All exceptions are caught and logged inside ``poll_device``; the function + never re-raises. This prevents a single broken device from aborting the + sweep for all other devices. +- ``poll_all_enabled_devices`` respects per-device ``poll_interval_s``: it skips + a device whose ``last_poll_at`` is recent enough. This lets the APScheduler + job run on a short base tick (e.g. 5 s) while each device self-regulates its + own cadence. +- The global ``modbus_polling_enabled`` flag from ``Settings`` (merged with any + DB overrides via ``build_runtime_settings``) gates the entire sweep as a + no-op. T08 will expose this flag in CONFIG_FIELDS/UI; for now it is + controlled by the ``MODBUS_POLLING_ENABLED`` env var (or the default True). +""" + +from __future__ import annotations + +import logging +from datetime import UTC, datetime + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.config import Settings +from app.integrations.modbus import driver, profiles +from app.models.modbus import ModbusDevice, ModbusReading +from app.services.config_page import build_runtime_settings + +logger = logging.getLogger(__name__) + +# Base polling tick in seconds (used by APScheduler; each device checks its own interval). +BASE_POLL_TICK_SECONDS = 5 + + +def poll_device(session: Session, device: ModbusDevice) -> ModbusReading | None: + """Read one enabled Modbus device, decode, and persist a reading row. + + Parameters + ---------- + session: + Open SQLAlchemy session. The caller is responsible for session + lifecycle (open / commit / close). + device: + ``ModbusDevice`` ORM instance to poll. + + Returns + ------- + ModbusReading | None + The newly-persisted reading on success, or ``None`` on any failure. + Failure is logged and the device's ``last_poll_ok`` is set to ``False``. + """ + now = datetime.now(UTC) + try: + profile = profiles.load_profile(device.profile) + registers = driver.read_blocks( + device.host, + device.port, + device.unit_id, + [{"start": b.start, "count": b.count} for b in profile.blocks], + ) + payload = profiles.decode(profile, registers) + + reading = ModbusReading( + device_id=device.id, + recorded_at=now, + payload=payload, + ) + session.add(reading) + + device.last_poll_at = now + device.last_poll_ok = True + session.commit() + + logger.debug( + "Polled device %r (id=%d): %d metrics recorded at %s", + device.friendly_name, + device.id, + len(payload), + now.isoformat(), + ) + return reading + + except Exception: # noqa: BLE001 + logger.exception( + "Failed to poll device %r (id=%d, host=%s:%d, unit_id=%d)", + device.friendly_name, + device.id, + device.host, + device.port, + device.unit_id, + ) + # Rollback first — if session.add(reading) already ran in the success + # branch before commit() failed, that pending ModbusReading must be + # discarded. Without rollback it would remain staged and could be + # flushed/committed by a subsequent operation (even for a different + # device), producing a spurious reading row for a poll that failed. + # Rollback also expires all ORM objects; device fields are re-set below. + try: + session.rollback() + except Exception: # noqa: BLE001 + logger.exception( + "session.rollback() failed for device id=%d; session may be unusable", device.id + ) + try: + device.last_poll_at = now + device.last_poll_ok = False + session.commit() + except Exception: # noqa: BLE001 + logger.exception( + "Also failed to persist last_poll_ok=False for device id=%d", device.id + ) + return None + + +def poll_all_enabled_devices( + session: Session, + *, + bootstrap_settings: Settings, +) -> None: + """Sweep all enabled Modbus devices and poll each one whose interval has elapsed. + + Global kill-switch: if ``build_runtime_settings(...).modbus_polling_enabled`` + is ``False``, this function returns immediately without touching any device. + + Per-device interval: a device is skipped if its ``last_poll_at`` is not yet + ``poll_interval_s`` seconds in the past. This allows the APScheduler job to + run on a short fixed tick while each device self-regulates its own cadence. + + Exceptions from individual devices are swallowed inside ``poll_device``; + this function itself will not raise. + + Parameters + ---------- + session: + Open SQLAlchemy session. + bootstrap_settings: + The bootstrap ``Settings`` instance (from ``get_settings()``), used as + the base for ``build_runtime_settings`` to pick up any DB overrides. + """ + try: + runtime_settings = build_runtime_settings(session, bootstrap_settings) + except Exception: # noqa: BLE001 + logger.exception("Failed to build runtime settings for Modbus poll; aborting sweep") + return + + if not runtime_settings.modbus_polling_enabled: + logger.debug("Modbus polling is disabled (modbus_polling_enabled=False); skipping sweep") + return + + try: + devices = session.execute( + select(ModbusDevice).where(ModbusDevice.enabled == True) # noqa: E712 + ).scalars().all() + except Exception: # noqa: BLE001 + logger.exception("Failed to query enabled Modbus devices; aborting sweep") + return + + now = datetime.now(UTC) + for device in devices: + # Respect per-device poll interval: skip if last poll was too recent. + if device.last_poll_at is not None: + # SQLite may return a naive datetime despite DateTime(timezone=True) — + # treat naive values as UTC so the comparison is safe. + last_poll = device.last_poll_at + if last_poll.tzinfo is None: + last_poll = last_poll.replace(tzinfo=UTC) + elapsed = (now - last_poll).total_seconds() + if elapsed < device.poll_interval_s: + logger.debug( + "Skipping device %r (id=%d): %.1fs elapsed < %ds interval", + device.friendly_name, + device.id, + elapsed, + device.poll_interval_s, + ) + continue + + poll_device(session, device) diff --git a/docs/design/m5-iot-energy.md b/docs/design/m5-iot-energy.md index 2e72163..0ddc258 100644 --- a/docs/design/m5-iot-energy.md +++ b/docs/design/m5-iot-energy.md @@ -329,7 +329,7 @@ Phase C（MQTT / Discovery，依赖 B 的设备数据与 provider 接口） - **Reviewer checklist**: 解码确为大端、高寄存器在前；地址与参考文档一致；**CLI 与 driver 都无任何写寄存器路径（仅 FC03/04）**；profile 纯协议知识、无部署项；`requirements.txt` 已同步锁定 `pymodbus`、`PyYAML`。 ### M5-T04 — 采集 service + APScheduler 轮询 -- **Status**: `todo` · **Depends**: M5-T03 +- **Status**: `done` · **Depends**: M5-T03 - **Context**: 周期扫所有 enabled 设备，调用 driver 读+解码，落 `modbus_reading.payload`。仿 public-ip 的同步 job 模式。 - **Files**: `create app/services/modbus_poll.py`；`modify app/main.py`（lifespan 注册 job）；`create tests/test_modbus_poll.py` - **Steps**: diff --git a/tests/test_modbus_poll.py b/tests/test_modbus_poll.py new file mode 100644 index 0000000..1a996a2 --- /dev/null +++ b/tests/test_modbus_poll.py @@ -0,0 +1,465 @@ +"""Tests for M5-T04: Modbus polling service (app/services/modbus_poll.py). + +Coverage: + 1. poll_all_enabled_devices inserts +N reading rows with correct payload + when the driver returns known register data. + 2. A device whose driver raises leaves last_poll_ok=False; other devices + in the same sweep are still polled and their readings are stored. + 3. modbus_polling_enabled=False causes the entire sweep to be skipped + (no readings inserted, no driver calls made). + 4. Per-device interval: a device whose last_poll_at is too recent is + skipped; one whose interval has elapsed is polled. + 5. poll_device returns None on driver failure and does not propagate. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session + +from app.config import Settings +from app.models.modbus import ModbusDevice, ModbusReading +from app.services.modbus_poll import poll_all_enabled_devices, poll_device + + +# --------------------------------------------------------------------------- +# Helpers and fixtures +# --------------------------------------------------------------------------- + + +def _make_app_alembic_config(database_url: str) -> Config: + cfg = Config("alembic_app.ini") + cfg.set_main_option("sqlalchemy.url", database_url) + return cfg + + +@pytest.fixture() +def poll_db(tmp_path: Path): + """Temporary SQLite DB upgraded to Alembic head, yields (engine, session_factory).""" + db_path = tmp_path / "poll_test.db" + db_url = f"sqlite:///{db_path}" + command.upgrade(_make_app_alembic_config(db_url), "head") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + yield engine + engine.dispose() + + +def _make_device( + session: Session, + *, + friendly_name: str = "Test Meter", + host: str = "10.0.0.1", + port: int = 502, + unit_id: int = 1, + profile: str = "sdm120", + poll_interval_s: int = 5, + enabled: bool = True, + last_poll_at: datetime | None = None, + last_poll_ok: bool | None = None, +) -> ModbusDevice: + now = datetime.now(UTC) + device = ModbusDevice( + friendly_name=friendly_name, + host=host, + port=port, + unit_id=unit_id, + profile=profile, + poll_interval_s=poll_interval_s, + enabled=enabled, + last_poll_at=last_poll_at, + last_poll_ok=last_poll_ok, + created_at=now, + updated_at=now, + ) + session.add(device) + session.commit() + return device + + +# Known payload that matches what decode() should produce for a minimal SDM120 profile. +KNOWN_PAYLOAD: dict[str, float] = { + "voltage": 230.20001220703125, + "current": 1.2999999523162842, + "active_power": 295.0, + "power_factor": 0.9800000190734863, + "frequency": 50.0, + "import_energy": 123.4000015258789, + "export_energy": 0.0, + "total_energy": 123.4000015258789, +} + + +def _make_bootstrap_settings(**overrides) -> Settings: + """Return a Settings instance with test defaults and optional overrides.""" + return Settings( + _env_file=None, + app_database_url="sqlite:///:memory:", + auth_bootstrap_password="test", + **overrides, + ) + + +# --------------------------------------------------------------------------- +# 1. Successful poll inserts +N readings with correct payload +# --------------------------------------------------------------------------- + + +def test_poll_all_inserts_readings_for_enabled_devices(poll_db): + """poll_all_enabled_devices inserts one reading per enabled device with correct payload.""" + with Session(poll_db) as session: + d1 = _make_device(session, friendly_name="Meter A", host="10.0.0.1") + d2 = _make_device(session, friendly_name="Meter B", host="10.0.0.2") + + # Verify baseline: no readings yet. + count_before = session.execute(select(ModbusReading)).scalars().all() + assert len(count_before) == 0 + + bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True) + + with ( + patch("app.services.modbus_poll.profiles.load_profile") as mock_load, + patch("app.services.modbus_poll.driver.read_blocks") as mock_read, + patch("app.services.modbus_poll.profiles.decode") as mock_decode, + ): + mock_profile = MagicMock() + mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)] + mock_load.return_value = mock_profile + mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334} + mock_decode.return_value = KNOWN_PAYLOAD + + poll_all_enabled_devices(session, bootstrap_settings=bootstrap) + + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 2, f"Expected 2 readings, got {len(readings)}" + + # Both readings should have the known payload. + for reading in readings: + assert reading.payload == KNOWN_PAYLOAD, ( + f"Payload mismatch for reading id={reading.id}: {reading.payload}" + ) + assert reading.recorded_at is not None + + # Devices should be marked last_poll_ok=True. + session.expire_all() + fetched_d1 = session.get(ModbusDevice, d1.id) + fetched_d2 = session.get(ModbusDevice, d2.id) + assert fetched_d1.last_poll_ok is True + assert fetched_d2.last_poll_ok is True + assert fetched_d1.last_poll_at is not None + assert fetched_d2.last_poll_at is not None + + +# --------------------------------------------------------------------------- +# 2. One device fails — others still succeed; failed device gets last_poll_ok=False +# --------------------------------------------------------------------------- + + +def test_poll_all_isolates_device_failure(poll_db): + """A driver error for one device must not prevent other devices from being polled.""" + with Session(poll_db) as session: + d_ok = _make_device(session, friendly_name="OK Meter", host="10.0.0.1") + d_fail = _make_device(session, friendly_name="Broken Meter", host="10.0.0.2") + + bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True) + + call_count = {"n": 0} + + def fake_read_blocks(host, port, unit_id, blocks, **kwargs): + call_count["n"] += 1 + if host == "10.0.0.2": + raise ConnectionError("Cannot connect to broken meter") + return {0x0000: 0x4366, 0x0001: 0x3334} + + with ( + patch("app.services.modbus_poll.profiles.load_profile") as mock_load, + patch("app.services.modbus_poll.driver.read_blocks", side_effect=fake_read_blocks), + patch("app.services.modbus_poll.profiles.decode") as mock_decode, + ): + mock_profile = MagicMock() + mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)] + mock_load.return_value = mock_profile + mock_decode.return_value = KNOWN_PAYLOAD + + # Must not raise even though one device fails. + poll_all_enabled_devices(session, bootstrap_settings=bootstrap) + + # Only the OK meter should have a reading. + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 1, f"Expected 1 reading (OK meter only), got {len(readings)}" + assert readings[0].device_id == d_ok.id + + # driver was called for both devices (fail attempt included). + assert call_count["n"] == 2 + + session.expire_all() + fetched_ok = session.get(ModbusDevice, d_ok.id) + fetched_fail = session.get(ModbusDevice, d_fail.id) + assert fetched_ok.last_poll_ok is True + assert fetched_fail.last_poll_ok is False + + +# --------------------------------------------------------------------------- +# 3. modbus_polling_enabled=False → entire sweep is skipped +# --------------------------------------------------------------------------- + + +def test_poll_all_skipped_when_polling_disabled(poll_db): + """When modbus_polling_enabled=False, no devices are polled and no readings are inserted.""" + with Session(poll_db) as session: + _make_device(session, friendly_name="Should Be Skipped", host="10.0.0.1") + + bootstrap = _make_bootstrap_settings(modbus_polling_enabled=False) + + with patch("app.services.modbus_poll.driver.read_blocks") as mock_read: + poll_all_enabled_devices(session, bootstrap_settings=bootstrap) + mock_read.assert_not_called() + + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 0, "No readings should be inserted when polling is disabled" + + +def test_poll_all_skipped_via_env_var(poll_db, monkeypatch): + """MODBUS_POLLING_ENABLED=false env var disables polling via Settings override.""" + monkeypatch.setenv("MODBUS_POLLING_ENABLED", "false") + # Build settings after env var is set. + bootstrap = Settings(_env_file=None, app_database_url="sqlite:///:memory:") + + assert bootstrap.modbus_polling_enabled is False, ( + "MODBUS_POLLING_ENABLED=false should yield modbus_polling_enabled=False" + ) + + with Session(poll_db) as session: + _make_device(session, friendly_name="Env-gated Meter", host="10.0.0.1") + + with patch("app.services.modbus_poll.driver.read_blocks") as mock_read: + poll_all_enabled_devices(session, bootstrap_settings=bootstrap) + mock_read.assert_not_called() + + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 0 + + +# --------------------------------------------------------------------------- +# 4. Per-device interval: skip if recently polled, poll if interval elapsed +# --------------------------------------------------------------------------- + + +def test_poll_all_respects_per_device_interval(poll_db): + """Devices polled more recently than poll_interval_s are skipped; others are polled.""" + now = datetime.now(UTC) + + with Session(poll_db) as session: + # Device polled just 1 second ago with a 30-second interval → should be SKIPPED. + d_recent = _make_device( + session, + friendly_name="Recent Meter", + host="10.0.0.1", + poll_interval_s=30, + last_poll_at=now - timedelta(seconds=1), + ) + # Device polled 60 seconds ago with a 30-second interval → should be POLLED. + d_due = _make_device( + session, + friendly_name="Due Meter", + host="10.0.0.2", + poll_interval_s=30, + last_poll_at=now - timedelta(seconds=60), + ) + # Device never polled before → should be POLLED. + d_new = _make_device( + session, + friendly_name="New Meter", + host="10.0.0.3", + poll_interval_s=30, + last_poll_at=None, + ) + + bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True) + + with ( + patch("app.services.modbus_poll.profiles.load_profile") as mock_load, + patch("app.services.modbus_poll.driver.read_blocks") as mock_read, + patch("app.services.modbus_poll.profiles.decode") as mock_decode, + ): + mock_profile = MagicMock() + mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)] + mock_load.return_value = mock_profile + mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334} + mock_decode.return_value = {"voltage": 230.2} + + poll_all_enabled_devices(session, bootstrap_settings=bootstrap) + + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 2, ( + f"Expected 2 readings (due + new), got {len(readings)}: " + f"{[r.device_id for r in readings]}" + ) + polled_device_ids = {r.device_id for r in readings} + assert d_due.id in polled_device_ids, "Due meter should have been polled" + assert d_new.id in polled_device_ids, "New meter should have been polled" + assert d_recent.id not in polled_device_ids, "Recent meter should have been skipped" + + +# --------------------------------------------------------------------------- +# 5. poll_device returns None on failure, does not propagate exception +# --------------------------------------------------------------------------- + + +def test_poll_device_returns_none_on_driver_failure(poll_db): + """poll_device swallows driver exceptions and returns None.""" + with Session(poll_db) as session: + device = _make_device(session, friendly_name="Failing Device", host="10.0.0.9") + + with ( + patch("app.services.modbus_poll.profiles.load_profile") as mock_load, + patch("app.services.modbus_poll.driver.read_blocks") as mock_read, + ): + mock_profile = MagicMock() + mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)] + mock_load.return_value = mock_profile + mock_read.side_effect = RuntimeError("Simulated driver failure") + + # Must NOT raise. + result = poll_device(session, device) + + assert result is None, "poll_device should return None on failure" + + session.expire_all() + fetched = session.get(ModbusDevice, device.id) + assert fetched.last_poll_ok is False + assert fetched.last_poll_at is not None + + # No reading row was created. + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 0 + + +def test_poll_device_returns_reading_on_success(poll_db): + """poll_device returns the inserted ModbusReading on success.""" + with Session(poll_db) as session: + device = _make_device(session, friendly_name="Success Device", host="10.0.0.5") + + with ( + patch("app.services.modbus_poll.profiles.load_profile") as mock_load, + patch("app.services.modbus_poll.driver.read_blocks") as mock_read, + patch("app.services.modbus_poll.profiles.decode") as mock_decode, + ): + mock_profile = MagicMock() + mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)] + mock_load.return_value = mock_profile + mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334} + mock_decode.return_value = KNOWN_PAYLOAD + + result = poll_device(session, device) + + assert result is not None, "poll_device should return a ModbusReading on success" + assert isinstance(result, ModbusReading) + assert result.device_id == device.id + assert result.payload == KNOWN_PAYLOAD + + session.expire_all() + fetched = session.get(ModbusDevice, device.id) + assert fetched.last_poll_ok is True + + +# --------------------------------------------------------------------------- +# 6. Disabled devices are not polled +# --------------------------------------------------------------------------- + + +def test_poll_all_skips_disabled_devices(poll_db): + """Devices with enabled=False must not be polled.""" + with Session(poll_db) as session: + _make_device(session, friendly_name="Disabled Meter", host="10.0.0.1", enabled=False) + + bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True) + + with patch("app.services.modbus_poll.driver.read_blocks") as mock_read: + poll_all_enabled_devices(session, bootstrap_settings=bootstrap) + mock_read.assert_not_called() + + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 0, "Disabled devices must not produce readings" + + +# --------------------------------------------------------------------------- +# 7. Rollback on commit failure: no reading survives a failed poll +# --------------------------------------------------------------------------- + + +def test_poll_device_commit_failure_leaves_no_reading(poll_db): + """When session.commit() raises after session.add(reading), the reading must NOT persist. + + Scenario (cross-device): two devices, device A's success-branch commit fails, + device B's poll succeeds normally. Expected outcome: + - 1 reading row total (device B only). + - device A: last_poll_ok=False, no reading row. + - device B: last_poll_ok=True, 1 reading row. + + This guards against the "zombie reading" bug where a pending ModbusReading + added in the success branch (before commit failed) survived in the session + and was flushed by device B's subsequent commit. + """ + with Session(poll_db) as session: + d_fail = _make_device(session, friendly_name="Commit-Fail Meter", host="10.0.0.1") + d_ok = _make_device(session, friendly_name="OK Meter", host="10.0.0.2") + + bootstrap = _make_bootstrap_settings(modbus_polling_enabled=True) + + # Track how many times commit() is called so we can make only the + # first call (device A's success-branch commit) raise while letting + # the subsequent calls (device A's failure-state commit, device B's + # success commit) succeed normally. + original_commit = session.commit + commit_call_count = {"n": 0} + + def patched_commit(): + commit_call_count["n"] += 1 + if commit_call_count["n"] == 1: + raise RuntimeError("Simulated commit failure (DB write error)") + return original_commit() + + with ( + patch("app.services.modbus_poll.profiles.load_profile") as mock_load, + patch("app.services.modbus_poll.driver.read_blocks") as mock_read, + patch("app.services.modbus_poll.profiles.decode") as mock_decode, + patch.object(session, "commit", side_effect=patched_commit), + ): + mock_profile = MagicMock() + mock_profile.blocks = [MagicMock(start=0x0000, count=0x0060)] + mock_load.return_value = mock_profile + mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334} + mock_decode.return_value = KNOWN_PAYLOAD + + # Must not raise despite the first commit failing. + poll_all_enabled_devices(session, bootstrap_settings=bootstrap) + + # Only device B (OK Meter) should have a reading — device A's reading + # must have been rolled back and must NOT have been committed via + # device B's subsequent commit (the "zombie reading" scenario). + readings = session.execute(select(ModbusReading)).scalars().all() + assert len(readings) == 1, ( + f"Expected exactly 1 reading (OK Meter only), got {len(readings)}: " + f"device_ids={[r.device_id for r in readings]}" + ) + assert readings[0].device_id == d_ok.id, ( + "The surviving reading must belong to the OK Meter, not the commit-fail device" + ) + + session.expire_all() + fetched_fail = session.get(ModbusDevice, d_fail.id) + fetched_ok = session.get(ModbusDevice, d_ok.id) + + assert fetched_fail.last_poll_ok is False, ( + "Commit-fail device must be marked last_poll_ok=False" + ) + assert fetched_ok.last_poll_ok is True, ( + "OK Meter must be marked last_poll_ok=True" + )