ems/backend/tests/test_modbus_execute_failsafe.py

"""execute_modbus_commands: žádná cesta nesmí nechat příkaz 'pending'.

Regrese na živý incident home-01 (TeltoCharge 172.16.1.16): zápisová trojice
(15, 19–20) buď skončila 'failed' s prázdným error_msg (str(TimeoutError())
== ''), nebo zůstala trvale 'pending' (export visel bez limitu na flock brány
obsazené pollingem mrtvého unit_id; výjimka mimo retry cyklus stav neuložila).

Testy: (1) error_msg nikdy prázdný; (2) GatewayLockTimeout → failed
s 'gateway lock timeout'; (3) CancelledError / chyba DB → safety net označí
zbylé příkazy failed a výjimku propaguje; (4) flock s timeoutem v
modbus_client; (5) backoff pollingu nedosažitelného wallboxu.
"""

import asyncio
import fcntl
import os
import tempfile
import unittest
from unittest.mock import AsyncMock, patch

import services.control.modbus_journal as journal
import services.modbus_client as mc
import services.telemetry_collector as tc
from services.control.modbus_journal import (
    _modbus_error_text,
    execute_modbus_commands,
)
from services.modbus_client import GatewayLockTimeout


def _cmd_row(cid: int, reg: int, val: int = 0) -> dict:
    return {
        "id": cid,
        "register": reg,
        "value_to_write": val,
        "device_host": "172.16.1.16",
        "device_port": 502,
        "device_unit_id": 1,
        "asset_code": "ev-charger-1",
    }


class _JournalDB:
    """In-memory journal — sleduje status a error_msg per command id."""

    def __init__(self, rows: list[dict], fail_written_update: bool = False) -> None:
        self.rows = {r["id"]: dict(r) for r in rows}
        self.status = {r["id"]: "pending" for r in rows}
        self.error_msg: dict[int, str | None] = {r["id"]: None for r in rows}
        self.fail_written_update = fail_written_update

    async def fetchrow(self, query: str, cid: int) -> dict | None:
        return self.rows.get(cid)

    async def execute(self, query: str, *args: object) -> None:
        if "status='written'" in query:
            if self.fail_written_update:
                raise RuntimeError("db connection lost")
            _val, cid = args
            self.status[int(cid)] = "written"  # type: ignore[arg-type]
            self.error_msg[int(cid)] = None  # type: ignore[arg-type]
        elif "status='failed'" in query:
            msg, cid = args
            self.status[int(cid)] = "failed"  # type: ignore[arg-type]
            self.error_msg[int(cid)] = str(msg)  # type: ignore[arg-type]
        else:
            raise AssertionError(f"unexpected execute: {query}")


def _fake_client(write_exc: BaseException | None = None) -> AsyncMock:
    client = AsyncMock()
    if write_exc is not None:
        client.write_registers.side_effect = write_exc
    client.force_disconnect = AsyncMock()
    return client


class ErrorTextTests(unittest.TestCase):
    def test_empty_str_exception_falls_back_to_repr(self) -> None:
        self.assertEqual(_modbus_error_text(TimeoutError()), "TimeoutError()")

    def test_nonempty_str_kept(self) -> None:
        self.assertEqual(_modbus_error_text(OSError("boom")), "boom")


class ExecuteFailsafeTests(unittest.IsolatedAsyncioTestCase):
    async def _run(
        self,
        db: _JournalDB,
        client: AsyncMock,
        ids: list[int],
    ) -> bool:
        with (
            patch.object(journal, "get_modbus_client", AsyncMock(return_value=client)),
            patch.object(journal.asyncio, "sleep", AsyncMock()),
        ):
            return await execute_modbus_commands(ids, db)  # type: ignore[arg-type]

    async def test_timeout_with_empty_str_marks_failed_with_nonempty_msg(self) -> None:
        db = _JournalDB([_cmd_row(1, 15), _cmd_row(2, 19), _cmd_row(3, 20)])
        ok = await self._run(db, _fake_client(TimeoutError()), [1, 2, 3])
        self.assertFalse(ok)
        self.assertEqual(set(db.status.values()), {"failed"})
        for msg in db.error_msg.values():
            self.assertTrue(msg)  # nikdy NULL/prázdný

    async def test_gateway_lock_timeout_marks_failed_with_reason(self) -> None:
        db = _JournalDB([_cmd_row(1, 15)])
        exc = GatewayLockTimeout("gateway lock timeout 172.16.1.16:502 after 20s")
        ok = await self._run(db, _fake_client(exc), [1])
        self.assertFalse(ok)
        self.assertEqual(db.status[1], "failed")
        self.assertIn("gateway lock timeout", db.error_msg[1] or "")

    async def test_cancelled_error_marks_failed_and_reraises(self) -> None:
        db = _JournalDB([_cmd_row(1, 15), _cmd_row(2, 19), _cmd_row(3, 20)])
        with self.assertRaises(asyncio.CancelledError):
            await self._run(db, _fake_client(asyncio.CancelledError()), [1, 2, 3])
        self.assertEqual(set(db.status.values()), {"failed"})
        for msg in db.error_msg.values():
            self.assertIn("execute aborted", msg or "")

    async def test_db_failure_in_written_update_marks_rest_failed(self) -> None:
        db = _JournalDB([_cmd_row(1, 15), _cmd_row(2, 19)], fail_written_update=True)
        with self.assertRaises(RuntimeError):
            await self._run(db, _fake_client(), [1, 2])
        self.assertEqual(set(db.status.values()), {"failed"})
        self.assertIn("db connection lost", db.error_msg[1] or "")

    async def test_force_disconnect_failure_does_not_leave_pending(self) -> None:
        db = _JournalDB([_cmd_row(1, 15)])
        client = _fake_client(OSError("write boom"))
        client.force_disconnect.side_effect = OSError("disconnect boom")
        ok = await self._run(db, client, [1])
        self.assertFalse(ok)
        self.assertEqual(db.status[1], "failed")
        self.assertIn("write boom", db.error_msg[1] or "")

    async def test_success_path_still_written(self) -> None:
        db = _JournalDB([_cmd_row(1, 15), _cmd_row(2, 19), _cmd_row(3, 20)])
        ok = await self._run(db, _fake_client(), [1, 2, 3])
        self.assertTrue(ok)
        self.assertEqual(set(db.status.values()), {"written"})


class GatewayFlockTimeoutTests(unittest.IsolatedAsyncioTestCase):
    async def test_lock_timeout_raises_gateway_lock_timeout(self) -> None:
        with tempfile.TemporaryDirectory() as d, patch.dict(
            os.environ,
            {"EMS_MODBUS_LOCK_DIR": d, "EMS_MODBUS_FLOCK_TIMEOUT_S": "0.3"},
        ):
            path = mc._gateway_lock_path("10.99.99.99", 502)
            path.parent.mkdir(parents=True, exist_ok=True)
            holder = open(path, "a+b")  # noqa: SIM115
            fcntl.flock(holder.fileno(), fcntl.LOCK_EX)
            try:
                with self.assertRaises(GatewayLockTimeout) as ctx:
                    async with mc._gateway_exclusive("10.99.99.99", 502):
                        pass
                self.assertIn("gateway lock timeout", str(ctx.exception))
            finally:
                fcntl.flock(holder.fileno(), fcntl.LOCK_UN)
                holder.close()

    async def test_lock_acquired_when_free(self) -> None:
        with tempfile.TemporaryDirectory() as d, patch.dict(
            os.environ, {"EMS_MODBUS_LOCK_DIR": d}
        ):
            async with mc._gateway_exclusive("10.99.99.98", 502):
                pass  # bez výjimky


class EvPollBackoffTests(unittest.TestCase):
    KEY = ("172.16.1.16", 502, 2)

    def setUp(self) -> None:
        tc._EV_POLL_FAIL_STREAK.clear()
        tc._EV_POLL_NEXT_ATTEMPT.clear()

    def test_below_threshold_never_skips(self) -> None:
        tc._ev_poll_record_failure(self.KEY, 100.0)
        tc._ev_poll_record_failure(self.KEY, 160.0)
        self.assertFalse(tc._ev_poll_should_skip(self.KEY, 220.0))

    def test_skips_after_threshold_until_backoff_elapses(self) -> None:
        for t in (100.0, 160.0, 220.0):
            tc._ev_poll_record_failure(self.KEY, t)
        self.assertTrue(tc._ev_poll_should_skip(self.KEY, 221.0))
        self.assertTrue(
            tc._ev_poll_should_skip(self.KEY, 220.0 + tc.EV_POLL_BACKOFF_S - 1)
        )
        self.assertFalse(
            tc._ev_poll_should_skip(self.KEY, 220.0 + tc.EV_POLL_BACKOFF_S + 1)
        )

    def test_success_resets_streak(self) -> None:
        for t in (100.0, 160.0, 220.0):
            tc._ev_poll_record_failure(self.KEY, t)
        tc._ev_poll_record_success(self.KEY)
        self.assertFalse(tc._ev_poll_should_skip(self.KEY, 221.0))


class _PollDB:
    """Jen řádek chargeru pro poll_ev_chargers (failure path se dál nedotkne DB)."""

    def __init__(self) -> None:
        self.row = {
            "id": 7,
            "code": "ev-charger-2",
            "host": "172.16.1.16",
            "port": 502,
            "unit_id": 2,
        }

    async def fetch(self, query: str, *args: object) -> list[dict]:
        return [self.row]


class PollEvChargersBackoffIntegrationTests(unittest.IsolatedAsyncioTestCase):
    async def test_dead_unit_stops_hitting_gateway_after_threshold(self) -> None:
        tc._EV_POLL_FAIL_STREAK.clear()
        tc._EV_POLL_NEXT_ATTEMPT.clear()
        get_client = AsyncMock(side_effect=OSError("unit 2 unreachable"))
        with patch.object(tc, "get_modbus_client", get_client):
            for _ in range(tc.EV_POLL_FAIL_THRESHOLD):
                await tc.poll_ev_chargers(1, _PollDB())  # type: ignore[arg-type]
            self.assertEqual(get_client.await_count, tc.EV_POLL_FAIL_THRESHOLD)
            # další tick uvnitř backoff okna už na bránu nesahá
            await tc.poll_ev_chargers(1, _PollDB())  # type: ignore[arg-type]
            self.assertEqual(get_client.await_count, tc.EV_POLL_FAIL_THRESHOLD)


if __name__ == "__main__":
    unittest.main()