| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111 |
- """Smart-plug timestamps must be naive UTC (#2539 collateral).
- Every ``DateTime`` column in the smart-plug tables is naive, and Bambuddy's
- convention is that a naive column holds UTC. The smart-plug code wrote *aware*
- datetimes into them anyway. SQLite tolerates that — its bind processor reads the
- datetime's fields and drops the offset — so it went unnoticed for a long time.
- **asyncpg does not.** It raises ``DataError: invalid input for query argument``,
- which meant that on Postgres:
- * every energy snapshot capture raised, so the snapshot table stayed empty and
- the Statistics page's date-filtered energy figure was permanently zero;
- * every plug status poll raised on ``last_checked``.
- Postgres is the setup Bambuddy recommends for multi-printer installs, so this
- was not a corner. Both of these tests fail against the pre-#2539 code.
- """
- from __future__ import annotations
- from types import SimpleNamespace
- from unittest.mock import AsyncMock, MagicMock, patch
- import pytest
- from backend.app.services import smart_plug_manager as manager_module
- from backend.app.utils.local_time import to_naive_utc, utcnow_naive
- def test_utcnow_naive_carries_no_offset():
- now = utcnow_naive()
- assert now.tzinfo is None
- def test_to_naive_utc_converts_rather_than_truncates():
- """An aware datetime in another zone must be *converted* to UTC before the
- offset is dropped, not merely stripped — stripping 02:00+02:00 would record
- it as 02:00 UTC, an hour of energy attributed to the wrong day.
- """
- from datetime import datetime, timedelta, timezone
- berlin_summer = timezone(timedelta(hours=2))
- aware = datetime(2026, 7, 11, 2, 30, tzinfo=berlin_summer)
- naive = to_naive_utc(aware)
- assert naive.tzinfo is None
- assert naive == datetime(2026, 7, 11, 0, 30)
- def test_to_naive_utc_passes_through_naive_and_none():
- from datetime import datetime
- already = datetime(2026, 7, 11, 12, 0)
- assert to_naive_utc(already) is already
- assert to_naive_utc(None) is None
- @pytest.mark.asyncio
- async def test_energy_snapshot_is_stamped_with_a_naive_datetime():
- """The regression guard, and the one that actually reproduces the bug.
- Reproducing the real failure needs a live Postgres, which CI has no reason to
- run for this. So catch it one step earlier: intercept the row on its way into
- the session and assert the timestamp carries no offset. An aware value here is
- exactly what asyncpg rejects with DataError, and what SQLite quietly swallows —
- which is why nobody noticed for so long.
- A source scan was tried first and was worthless: the aware ``now`` is assigned
- on one line and used as ``recorded_at`` on another, so grepping for the two
- together sees nothing.
- """
- added: list[object] = []
- class FakeSession:
- async def execute(self, *_a, **_kw):
- result = MagicMock()
- result.scalars.return_value.all.return_value = [SimpleNamespace(id=1, plug_type="rest", enabled=True)]
- return result
- def add(self, obj):
- added.append(obj)
- async def commit(self):
- pass
- async def __aenter__(self):
- return self
- async def __aexit__(self, *_a):
- return False
- manager = manager_module.SmartPlugManager()
- with (
- patch("backend.app.core.database.async_session", FakeSession),
- patch.object(
- manager,
- "get_service_for_plug",
- new=AsyncMock(return_value=SimpleNamespace(get_energy=AsyncMock(return_value={"total": 2.62}))),
- ),
- ):
- await manager._capture_energy_snapshots()
- assert len(added) == 1, "expected one snapshot row to be written"
- recorded_at = added[0].recorded_at
- assert recorded_at.tzinfo is None, (
- "energy snapshot stamped with a timezone-aware datetime. The column is "
- "naive, and asyncpg raises DataError on Postgres — the whole capture "
- "fails and the Statistics energy figure stays at zero. Use utcnow_naive()."
- )
|