Spaces:

Catalna
/

ginchiee-bot

Running on Zero

fix: await the timezone lookup, and copy Postgres data across environments

7d3a67d 28 days ago

19.3 kB

"""
tests/test_database.py
Unit tests for the placeholder translator, plus an integration suite that runs
against a real PostgreSQL database when DATABASE_URL is available.
The integration tests are the point: they prove the wrapper keeps the aiosqlite
call patterns working (`await db.execute(...)`, `async with db.execute(...) as
cursor`), that `?` never breaks inside string literals, that migrations are
idempotent, and that transactional writes roll back.
Run:
    python -m unittest tests.test_database -v
"""

import asyncio
import os
import unittest
from pathlib import Path
import sys

PROJECT_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT_ROOT))

from database.connection import translate_placeholders  # noqa: E402

PROBE_USER_ID = 999_999_001


class PlaceholderTranslationTests(unittest.TestCase):
    """Pure functions — no database required."""

    def test_single_placeholder(self) -> None:
        self.assertEqual(
            translate_placeholders("SELECT * FROM users WHERE user_id = ?"),
            "SELECT * FROM users WHERE user_id = $1",
        )

    def test_multiple_placeholders_number_in_order(self) -> None:
        self.assertEqual(
            translate_placeholders("UPDATE t SET a = ?, b = ? WHERE id = ?"),
            "UPDATE t SET a = $1, b = $2 WHERE id = $3",
        )

    def test_question_mark_inside_string_literal_is_preserved(self) -> None:
        sql = "INSERT INTO t (q) VALUES ('sudah makan?')"
        result = translate_placeholders(sql)
        self.assertEqual(result, "INSERT INTO t (q) VALUES ('sudah makan?')")
        self.assertNotIn("$1", result)

    def test_mixed_placeholder_and_literal(self) -> None:
        sql = "SELECT * FROM t WHERE name = ? AND note = 'why?'"
        self.assertEqual(
            translate_placeholders(sql),
            "SELECT * FROM t WHERE name = $1 AND note = 'why?'",
        )

    def test_escaped_quote_in_literal(self) -> None:
        sql = "INSERT INTO t (v) VALUES ('it''s ok?')"
        self.assertEqual(
            translate_placeholders(sql),
            "INSERT INTO t (v) VALUES ('it''s ok?')",
        )

    def test_no_placeholder(self) -> None:
        sql = "SELECT now()"
        self.assertEqual(translate_placeholders(sql), sql)


def _resolve_database_url() -> str | None:
    """Read DATABASE_URL from the environment or the project .env file."""
    if os.getenv("DATABASE_URL"):
        return os.getenv("DATABASE_URL")
    env_path = PROJECT_ROOT / ".env"
    if not env_path.exists():
        return None
    for line in env_path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if line.startswith("DATABASE_URL="):
            value = line.split("=", 1)[1].strip()
            return value or None
    return None


class _FrozenDatetime:
    """Stand-in for `datetime` whose `now(tz)` returns a fixed instant.
    Used to pin the clock inside timezone assertions without touching the
    system clock (which would be flaky and affect other tests).
    """

    def __init__(self, moment):
        self._moment = moment

    def now(self, tz=None):
        if tz is None:
            return self._moment.replace(tzinfo=None)
        return self._moment.astimezone(tz)

    def date(self):
        return self._moment.date()


@unittest.skipUnless(_resolve_database_url(), "DATABASE_URL not available")
class PostgresIntegrationTests(unittest.TestCase):
    """All checks run inside one event loop (see module docstring)."""

    def setUp(self) -> None:
        from database.connection import close_db

        # A previous run may have left the pool bound to a dead loop; closing it
        # here is safe because the pool is torn down in the same loop that will
        # recreate it, at the end of _run_all_checks.
        self._close_db = close_db
        asyncio.run(self._close_pool_if_possible())

    async def _close_pool_if_possible(self) -> None:
        import database.connection as connection_module

        if connection_module._pool is not None:
            try:
                await connection_module._pool.close()
            except Exception:
                pass
            connection_module._pool = None

    # ── Individual checks ────────────────────────────────────────────────────

    async def _reset_probe(self, db) -> None:
        """Remove any probe rows so each check starts from a clean slate."""
        await db.execute(
            "DELETE FROM user_activities WHERE user_id = $1", (PROBE_USER_ID,)
        )
        await db.execute("DELETE FROM users WHERE user_id = $1", (PROBE_USER_ID,))

    async def _check_migrations_idempotent(self, db) -> None:
        from database.migrations import run_migrations

        await run_migrations()
        second = await run_migrations()
        self.assertEqual(second, [], f"re-running migrations applied {second}")

        recorded = await db.execute(
            "SELECT version, name FROM schema_migrations ORDER BY version"
        ).fetchall()
        self.assertTrue(recorded, "no migrations recorded")
        self.assertEqual(recorded[0]["version"], 1)

    async def _check_tables(self, db) -> None:
        rows = await db.execute(
            """
            SELECT table_name FROM information_schema.tables
            WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
            """
        ).fetchall()
        names = {row[0] for row in rows}

        expected = {
            "users", "diet_profiles", "meal_schedules", "food_logs",
            "conversations", "user_activities", "schema_migrations",
        }
        self.assertTrue(expected.issubset(names), f"missing tables: {sorted(expected - names)}")

        # google_calendar_tokens: orphaned, no code references it.
        # daily_nutrition: write-only, never read.
        self.assertNotIn("google_calendar_tokens", names)
        self.assertNotIn("daily_nutrition", names)

    async def _check_await_execute(self, db) -> None:
        await db.execute(
            """
            INSERT INTO users (user_id, name, username, timezone)
            VALUES ($1, $2, $3, $4)
            ON CONFLICT (user_id) DO UPDATE SET name = excluded.name
            """,
            (PROBE_USER_ID, "Test User", "tester", "Asia/Jakarta"),
        )
        await db.commit()

        value = await db.execute(
            "SELECT name FROM users WHERE user_id = $1", (PROBE_USER_ID,)
        ).fetchval()
        self.assertEqual(value, "Test User")

    async def _check_async_with_cursor(self, db) -> None:
        await db.execute(
            "INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
            (PROBE_USER_ID, "Cursor Test", "cur"),
        )
        async with db.execute(
            "SELECT user_id, name, timezone FROM users WHERE user_id = $1",
            (PROBE_USER_ID,),
        ) as cursor:
            row = await cursor.fetchone()
        self.assertIsNotNone(row)
        self.assertEqual(dict(row)["name"], "Cursor Test")
        self.assertEqual(row[0], PROBE_USER_ID)

    async def _check_returning_id(self, db) -> None:
        # Foreign keys are enforced, so the parent user must exist first.
        await db.execute(
            "INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
            (PROBE_USER_ID, "Returning Test", "ret"),
        )

        activity_id = await db.execute(
            """
            INSERT INTO user_activities (user_id, title, activity_dt, remind_mins)
            VALUES ($1, $2, now() + interval '1 day', 30)
            RETURNING id
            """,
            (PROBE_USER_ID, "Test Activity"),
        ).fetchval()
        self.assertIsInstance(activity_id, int)

        deleted = await db.execute(
            "DELETE FROM user_activities WHERE id = $1 AND user_id = $2 RETURNING id",
            (activity_id, PROBE_USER_ID),
        ).fetchone()
        self.assertIsNotNone(deleted)

    async def _check_meal_type_inference(self, db) -> None:
        """meal_type used to be permanently NULL, making /today emoji useless."""
        from services.nutrition_service import get_today_food_logs, user_today

        await db.execute(
            "INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
            (PROBE_USER_ID, "Meal Type Test", "mt"),
        )

        logged = await db.execute(
            """
            INSERT INTO food_logs (user_id, date, meal_type, food_name, amount_g,
                                   calories, protein_g, carbs_g, fat_g)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            RETURNING id
            """,
            (PROBE_USER_ID, await user_today(PROBE_USER_ID), "lunch", "nasi goreng",
             200.0, 320.0, 8.0, 45.0, 10.0),
        ).fetchval()

        self.assertIsInstance(logged, int)
        rows = await get_today_food_logs(PROBE_USER_ID)
        self.assertTrue(rows)
        self.assertEqual(rows[-1]["meal_type"], "lunch")

    async def _check_today_follows_user_timezone(self, db) -> None:
        """Regression test for a real data-correctness bug.
        date.today() read the machine's timezone. On Hugging Face the
        container runs in UTC while users are in Asia/Jakarta (UTC+7), so
        between 00:00 and 07:00 WIB the UTC date is still the previous day.
        Meals logged then got filed under yesterday and /today showed nothing.
        The previous version of this fix called the async `_tz_for()` without
        awaiting it, so the lookup silently failed and user_today() fell back
        to the machine timezone — meaning the test passed through the fallback.
        This test therefore proves the timezone is genuinely read from
        `users.timezone` by setting an UNUSUAL zone and expecting the date to
        follow it, which only happens if the query really runs.
        """
        from datetime import datetime
        from zoneinfo import ZoneInfo

        from services.nutrition_service import user_today
        from services.user_service import (
            create_or_update_user,
            update_user_timezone,
        )

        probe = PROBE_USER_ID + 41

        # WIT is UTC+9 and WIB is UTC+7, so only between 22:00 and 23:59 WIB do
        # the two zones fall on different calendar days.
        wib = ZoneInfo("Asia/Jakarta")
        wit = ZoneInfo("Asia/Jayapura")
        sample = datetime(2026, 10, 12, 23, 0, tzinfo=wib)  # 23:00 WIB

        expected_with_jakarta = sample.astimezone(wib).date()
        expected_with_jayapura = sample.astimezone(wit).date()

        if expected_with_jayapura == expected_with_jakarta:
            self.skipTest("sample date lands on the same day in both zones")

        await db.execute("DELETE FROM users WHERE user_id=$1", (probe,))
        await create_or_update_user(probe, "TZ Probe", None, "Asia/Jakarta")
        try:
            import services.nutrition_service as ns

            original = ns.datetime
            ns.datetime = _FrozenDatetime(sample)
            try:
                # With the user on WIB.
                await update_user_timezone(probe, "Asia/Jakarta")
                as_jakarta = await user_today(probe)

                # With the user on WIT (UTC+9) the SAME instant is a later date,
                # so only a real timezone lookup can produce this.
                await update_user_timezone(probe, "Asia/Jayapura")
                as_jayapura = await user_today(probe)
            finally:
                ns.datetime = original

            self.assertEqual(
                as_jakarta, expected_with_jakarta,
                "user_today must honour Asia/Jakarta",
            )
            self.assertEqual(
                as_jayapura, expected_with_jayapura,
                "user_today must change when the stored timezone changes "
                "(a fallback would keep returning the machine's date)",
            )
            self.assertNotEqual(
                as_jayapura, as_jakarta,
                "the two zones must actually differ for this test to mean anything",
            )
        finally:
            await db.execute("DELETE FROM users WHERE user_id=$1", (probe,))

    async def _check_activity_due_reminder(self, db) -> None:
        from datetime import datetime, timedelta, timezone

        from services.activity_service import (
            add_activity,
            get_due_activity_reminders,
            delete_activity,
        )

        await db.execute(
            "INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
            (PROBE_USER_ID, "Reminder Test", "rem"),
        )

        # 10 minutes away with a 30 minute lead time -> already due.
        soon = datetime.now(timezone.utc) + timedelta(minutes=10)
        # 3 days away with a 30 minute lead time -> not due.
        later = datetime.now(timezone.utc) + timedelta(days=3)

        soon_id = await add_activity(
            PROBE_USER_ID, "Reminder Soon", soon.isoformat(timespec="seconds"), remind_mins=30
        )
        later_id = await add_activity(
            PROBE_USER_ID, "Reminder Later", later.isoformat(timespec="seconds"), remind_mins=30
        )

        due = await get_due_activity_reminders()
        due_ids = {row["id"] for row in due}

        dump = await db.execute(
            """
            SELECT id, title, activity_dt, remind_mins, reminded, user_id
            FROM user_activities WHERE user_id = $1
            """,
            (PROBE_USER_ID,),
        ).fetchall()
        now_val = await db.execute("SELECT now()").fetchval()
        print(f"    [diag] now={now_val}")
        for r in dump:
            due_at = r[2].timestamp() and (r[2] - __import__("datetime").timedelta(minutes=r[3]))
            print(f"    [diag] row id={r[0]} title={r[1]!r} dt={r[2]} remind={r[3]} due_at={due_at} in_due={r[0] in due_ids}")
        print(f"    [diag] due_ids={sorted(due_ids)} soon={soon_id} later={later_id}")

        self.assertIn(soon_id, due_ids, "reminder within the lead window should be due")
        self.assertNotIn(later_id, due_ids, "activity 3 days out should not be due")

        await delete_activity(soon_id, PROBE_USER_ID)
        await delete_activity(later_id, PROBE_USER_ID)

    async def _check_remind_mins_is_bounded(self, db) -> None:
        """An absurd lead time must not make far-future activities fire now."""
        from datetime import datetime, timedelta, timezone

        from services.activity_service import (
            add_activity,
            get_due_activity_reminders,
            delete_activity,
        )

        await db.execute(
            "INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
            (PROBE_USER_ID, "Bounds Test", "bnd"),
        )

        far = datetime.now(timezone.utc) + timedelta(days=3)
        activity_id = await add_activity(
            PROBE_USER_ID, "Far Future", far.isoformat(timespec="seconds"),
            remind_mins=999_999_999,
        )
        try:
            due_ids = {row["id"] for row in await get_due_activity_reminders()}
            self.assertNotIn(
                activity_id, due_ids,
                "an absurd remind_mins must not fire a far-future activity",
            )
        finally:
            await delete_activity(activity_id, PROBE_USER_ID)

    async def _check_question_mark_translation(self, db) -> None:
        await db.execute(
            "INSERT INTO users (user_id, name, username) VALUES (?, ?, ?)",
            (PROBE_USER_ID, "Question Mark", "qm"),
        )
        async with db.execute(
            "SELECT name FROM users WHERE user_id = ?", (PROBE_USER_ID,)
        ) as cursor:
            row = await cursor.fetchone()
        self.assertEqual(row[0], "Question Mark")

    async def _check_literal_question_mark(self, db) -> None:
        await db.execute(
            "INSERT INTO users (user_id, name, username) VALUES (?, ?, ?)",
            (PROBE_USER_ID, "sudah makan?", "qmlit"),
        )
        value = await db.execute(
            "SELECT name FROM users WHERE user_id = ?", (PROBE_USER_ID,)
        ).fetchval()
        self.assertEqual(value, "sudah makan?")

    async def _check_transaction_rollback(self, db) -> None:
        async def _fail(tx) -> None:
            await tx.execute(
                "INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
                (PROBE_USER_ID, "Tx Rollback", "txrb"),
            )
            # Second insert violates the PRIMARY KEY on user_id.
            await tx.execute(
                "INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
                (PROBE_USER_ID, "Duplicate", "txdup"),
            )

        with self.assertRaises(Exception):
            async with db.transaction() as tx:
                await _fail(tx)

        exists = await db.execute(
            "SELECT 1 FROM users WHERE user_id = $1", (PROBE_USER_ID,)
        ).fetchone()
        self.assertIsNone(exists, "transaction was not rolled back")

    # ── Suite ────────────────────────────────────────────────────────────────

    def test_all_checks_against_real_database(self) -> None:
        asyncio.run(self._run_all_checks())

    async def _run_all_checks(self) -> None:
        from database.connection import init_db

        db = await init_db()

        checks = [
            ("migrations idempotent", self._check_migrations_idempotent),
            ("expected tables exist", self._check_tables),
            ("await db.execute pattern", self._check_await_execute),
            ("async with cursor pattern", self._check_async_with_cursor),
            ("RETURNING id + fetchval", self._check_returning_id),
            ("meal_type is populated", self._check_meal_type_inference),
            ("today follows user timezone", self._check_today_follows_user_timezone),
            ("? placeholder translation", self._check_question_mark_translation),
            ("literal ? preserved", self._check_literal_question_mark),
            ("transaction rollback", self._check_transaction_rollback),
            ("timestamptz due reminders", self._check_activity_due_reminder),
            ("remind_mins bounded", self._check_remind_mins_is_bounded),
        ]

        failures = []
        for name, check in checks:
            await self._reset_probe(db)
            try:
                await check(db)
                print(f"  PASS  {name}")
            except Exception as exc:  # noqa: BLE001
                print(f"  FAIL  {name}: {exc}")
                failures.append((name, exc))

        # Always clean the probe user so the next run starts clean.
        await self._reset_probe(db)

        await self._reset_probe(db)
        # Close in the same event loop that created the pool.
        from database.connection import close_db
        await close_db()

        if failures:
            self.fail(f"{len(failures)} check(s) failed: {[n for n, _ in failures]}")


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

	"""
	tests/test_database.py

	Unit tests for the placeholder translator, plus an integration suite that runs
	against a real PostgreSQL database when DATABASE_URL is available.

	The integration tests are the point: they prove the wrapper keeps the aiosqlite
	call patterns working (`await db.execute(...)`, `async with db.execute(...) as
	cursor`), that `?` never breaks inside string literals, that migrations are
	idempotent, and that transactional writes roll back.

	Run:
	python -m unittest tests.test_database -v
	"""

	import asyncio
	import os
	import unittest
	from pathlib import Path
	import sys

	PROJECT_ROOT = Path(__file__).resolve().parents[1]
	sys.path.insert(0, str(PROJECT_ROOT))

	from database.connection import translate_placeholders # noqa: E402

	PROBE_USER_ID = 999_999_001


	class PlaceholderTranslationTests(unittest.TestCase):
	"""Pure functions — no database required."""

	def test_single_placeholder(self) -> None:
	self.assertEqual(
	translate_placeholders("SELECT * FROM users WHERE user_id = ?"),
	"SELECT * FROM users WHERE user_id = $1",
	)

	def test_multiple_placeholders_number_in_order(self) -> None:
	self.assertEqual(
	translate_placeholders("UPDATE t SET a = ?, b = ? WHERE id = ?"),
	"UPDATE t SET a = $1, b = $2 WHERE id = $3",
	)

	def test_question_mark_inside_string_literal_is_preserved(self) -> None:
	sql = "INSERT INTO t (q) VALUES ('sudah makan?')"
	result = translate_placeholders(sql)
	self.assertEqual(result, "INSERT INTO t (q) VALUES ('sudah makan?')")
	self.assertNotIn("$1", result)

	def test_mixed_placeholder_and_literal(self) -> None:
	sql = "SELECT * FROM t WHERE name = ? AND note = 'why?'"
	self.assertEqual(
	translate_placeholders(sql),
	"SELECT * FROM t WHERE name = $1 AND note = 'why?'",
	)

	def test_escaped_quote_in_literal(self) -> None:
	sql = "INSERT INTO t (v) VALUES ('it''s ok?')"
	self.assertEqual(
	translate_placeholders(sql),
	"INSERT INTO t (v) VALUES ('it''s ok?')",
	)

	def test_no_placeholder(self) -> None:
	sql = "SELECT now()"
	self.assertEqual(translate_placeholders(sql), sql)


	def _resolve_database_url() -> str \| None:
	"""Read DATABASE_URL from the environment or the project .env file."""
	if os.getenv("DATABASE_URL"):
	return os.getenv("DATABASE_URL")
	env_path = PROJECT_ROOT / ".env"
	if not env_path.exists():
	return None
	for line in env_path.read_text(encoding="utf-8").splitlines():
	line = line.strip()
	if line.startswith("DATABASE_URL="):
	value = line.split("=", 1)[1].strip()
	return value or None
	return None


	class _FrozenDatetime:
	"""Stand-in for `datetime` whose `now(tz)` returns a fixed instant.

	Used to pin the clock inside timezone assertions without touching the
	system clock (which would be flaky and affect other tests).
	"""

	def __init__(self, moment):
	self._moment = moment

	def now(self, tz=None):
	if tz is None:
	return self._moment.replace(tzinfo=None)
	return self._moment.astimezone(tz)

	def date(self):
	return self._moment.date()


	@unittest.skipUnless(_resolve_database_url(), "DATABASE_URL not available")
	class PostgresIntegrationTests(unittest.TestCase):
	"""All checks run inside one event loop (see module docstring)."""

	def setUp(self) -> None:
	from database.connection import close_db

	# A previous run may have left the pool bound to a dead loop; closing it
	# here is safe because the pool is torn down in the same loop that will
	# recreate it, at the end of _run_all_checks.
	self._close_db = close_db
	asyncio.run(self._close_pool_if_possible())

	async def _close_pool_if_possible(self) -> None:
	import database.connection as connection_module

	if connection_module._pool is not None:
	try:
	await connection_module._pool.close()
	except Exception:
	pass
	connection_module._pool = None

	# ── Individual checks ────────────────────────────────────────────────────

	async def _reset_probe(self, db) -> None:
	"""Remove any probe rows so each check starts from a clean slate."""
	await db.execute(
	"DELETE FROM user_activities WHERE user_id = $1", (PROBE_USER_ID,)
	)
	await db.execute("DELETE FROM users WHERE user_id = $1", (PROBE_USER_ID,))

	async def _check_migrations_idempotent(self, db) -> None:
	from database.migrations import run_migrations

	await run_migrations()
	second = await run_migrations()
	self.assertEqual(second, [], f"re-running migrations applied {second}")

	recorded = await db.execute(
	"SELECT version, name FROM schema_migrations ORDER BY version"
	).fetchall()
	self.assertTrue(recorded, "no migrations recorded")
	self.assertEqual(recorded[0]["version"], 1)

	async def _check_tables(self, db) -> None:
	rows = await db.execute(
	"""
	SELECT table_name FROM information_schema.tables
	WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
	"""
	).fetchall()
	names = {row[0] for row in rows}

	expected = {
	"users", "diet_profiles", "meal_schedules", "food_logs",
	"conversations", "user_activities", "schema_migrations",
	}
	self.assertTrue(expected.issubset(names), f"missing tables: {sorted(expected - names)}")

	# google_calendar_tokens: orphaned, no code references it.
	# daily_nutrition: write-only, never read.
	self.assertNotIn("google_calendar_tokens", names)
	self.assertNotIn("daily_nutrition", names)

	async def _check_await_execute(self, db) -> None:
	await db.execute(
	"""
	INSERT INTO users (user_id, name, username, timezone)
	VALUES ($1, $2, $3, $4)
	ON CONFLICT (user_id) DO UPDATE SET name = excluded.name
	""",
	(PROBE_USER_ID, "Test User", "tester", "Asia/Jakarta"),
	)
	await db.commit()

	value = await db.execute(
	"SELECT name FROM users WHERE user_id = $1", (PROBE_USER_ID,)
	).fetchval()
	self.assertEqual(value, "Test User")

	async def _check_async_with_cursor(self, db) -> None:
	await db.execute(
	"INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
	(PROBE_USER_ID, "Cursor Test", "cur"),
	)
	async with db.execute(
	"SELECT user_id, name, timezone FROM users WHERE user_id = $1",
	(PROBE_USER_ID,),
	) as cursor:
	row = await cursor.fetchone()
	self.assertIsNotNone(row)
	self.assertEqual(dict(row)["name"], "Cursor Test")
	self.assertEqual(row[0], PROBE_USER_ID)

	async def _check_returning_id(self, db) -> None:
	# Foreign keys are enforced, so the parent user must exist first.
	await db.execute(
	"INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
	(PROBE_USER_ID, "Returning Test", "ret"),
	)

	activity_id = await db.execute(
	"""
	INSERT INTO user_activities (user_id, title, activity_dt, remind_mins)
	VALUES ($1, $2, now() + interval '1 day', 30)
	RETURNING id
	""",
	(PROBE_USER_ID, "Test Activity"),
	).fetchval()
	self.assertIsInstance(activity_id, int)

	deleted = await db.execute(
	"DELETE FROM user_activities WHERE id = $1 AND user_id = $2 RETURNING id",
	(activity_id, PROBE_USER_ID),
	).fetchone()
	self.assertIsNotNone(deleted)

	async def _check_meal_type_inference(self, db) -> None:
	"""meal_type used to be permanently NULL, making /today emoji useless."""
	from services.nutrition_service import get_today_food_logs, user_today

	await db.execute(
	"INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
	(PROBE_USER_ID, "Meal Type Test", "mt"),
	)

	logged = await db.execute(
	"""
	INSERT INTO food_logs (user_id, date, meal_type, food_name, amount_g,
	calories, protein_g, carbs_g, fat_g)
	VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
	RETURNING id
	""",
	(PROBE_USER_ID, await user_today(PROBE_USER_ID), "lunch", "nasi goreng",
	200.0, 320.0, 8.0, 45.0, 10.0),
	).fetchval()

	self.assertIsInstance(logged, int)
	rows = await get_today_food_logs(PROBE_USER_ID)
	self.assertTrue(rows)
	self.assertEqual(rows[-1]["meal_type"], "lunch")

	async def _check_today_follows_user_timezone(self, db) -> None:
	"""Regression test for a real data-correctness bug.

	date.today() read the machine's timezone. On Hugging Face the
	container runs in UTC while users are in Asia/Jakarta (UTC+7), so
	between 00:00 and 07:00 WIB the UTC date is still the previous day.
	Meals logged then got filed under yesterday and /today showed nothing.

	The previous version of this fix called the async `_tz_for()` without
	awaiting it, so the lookup silently failed and user_today() fell back
	to the machine timezone — meaning the test passed through the fallback.
	This test therefore proves the timezone is genuinely read from
	`users.timezone` by setting an UNUSUAL zone and expecting the date to
	follow it, which only happens if the query really runs.
	"""
	from datetime import datetime
	from zoneinfo import ZoneInfo

	from services.nutrition_service import user_today
	from services.user_service import (
	create_or_update_user,
	update_user_timezone,
	)

	probe = PROBE_USER_ID + 41

	# WIT is UTC+9 and WIB is UTC+7, so only between 22:00 and 23:59 WIB do
	# the two zones fall on different calendar days.
	wib = ZoneInfo("Asia/Jakarta")
	wit = ZoneInfo("Asia/Jayapura")
	sample = datetime(2026, 10, 12, 23, 0, tzinfo=wib) # 23:00 WIB

	expected_with_jakarta = sample.astimezone(wib).date()
	expected_with_jayapura = sample.astimezone(wit).date()

	if expected_with_jayapura == expected_with_jakarta:
	self.skipTest("sample date lands on the same day in both zones")

	await db.execute("DELETE FROM users WHERE user_id=$1", (probe,))
	await create_or_update_user(probe, "TZ Probe", None, "Asia/Jakarta")
	try:
	import services.nutrition_service as ns

	original = ns.datetime
	ns.datetime = _FrozenDatetime(sample)
	try:
	# With the user on WIB.
	await update_user_timezone(probe, "Asia/Jakarta")
	as_jakarta = await user_today(probe)

	# With the user on WIT (UTC+9) the SAME instant is a later date,
	# so only a real timezone lookup can produce this.
	await update_user_timezone(probe, "Asia/Jayapura")
	as_jayapura = await user_today(probe)
	finally:
	ns.datetime = original

	self.assertEqual(
	as_jakarta, expected_with_jakarta,
	"user_today must honour Asia/Jakarta",
	)
	self.assertEqual(
	as_jayapura, expected_with_jayapura,
	"user_today must change when the stored timezone changes "
	"(a fallback would keep returning the machine's date)",
	)
	self.assertNotEqual(
	as_jayapura, as_jakarta,
	"the two zones must actually differ for this test to mean anything",
	)
	finally:
	await db.execute("DELETE FROM users WHERE user_id=$1", (probe,))

	async def _check_activity_due_reminder(self, db) -> None:
	from datetime import datetime, timedelta, timezone

	from services.activity_service import (
	add_activity,
	get_due_activity_reminders,
	delete_activity,
	)

	await db.execute(
	"INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
	(PROBE_USER_ID, "Reminder Test", "rem"),
	)

	# 10 minutes away with a 30 minute lead time -> already due.
	soon = datetime.now(timezone.utc) + timedelta(minutes=10)
	# 3 days away with a 30 minute lead time -> not due.
	later = datetime.now(timezone.utc) + timedelta(days=3)

	soon_id = await add_activity(
	PROBE_USER_ID, "Reminder Soon", soon.isoformat(timespec="seconds"), remind_mins=30
	)
	later_id = await add_activity(
	PROBE_USER_ID, "Reminder Later", later.isoformat(timespec="seconds"), remind_mins=30
	)

	due = await get_due_activity_reminders()
	due_ids = {row["id"] for row in due}

	dump = await db.execute(
	"""
	SELECT id, title, activity_dt, remind_mins, reminded, user_id
	FROM user_activities WHERE user_id = $1
	""",
	(PROBE_USER_ID,),
	).fetchall()
	now_val = await db.execute("SELECT now()").fetchval()
	print(f" [diag] now={now_val}")
	for r in dump:
	due_at = r[2].timestamp() and (r[2] - __import__("datetime").timedelta(minutes=r[3]))
	print(f" [diag] row id={r[0]} title={r[1]!r} dt={r[2]} remind={r[3]} due_at={due_at} in_due={r[0] in due_ids}")
	print(f" [diag] due_ids={sorted(due_ids)} soon={soon_id} later={later_id}")

	self.assertIn(soon_id, due_ids, "reminder within the lead window should be due")
	self.assertNotIn(later_id, due_ids, "activity 3 days out should not be due")

	await delete_activity(soon_id, PROBE_USER_ID)
	await delete_activity(later_id, PROBE_USER_ID)

	async def _check_remind_mins_is_bounded(self, db) -> None:
	"""An absurd lead time must not make far-future activities fire now."""
	from datetime import datetime, timedelta, timezone

	from services.activity_service import (
	add_activity,
	get_due_activity_reminders,
	delete_activity,
	)

	await db.execute(
	"INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
	(PROBE_USER_ID, "Bounds Test", "bnd"),
	)

	far = datetime.now(timezone.utc) + timedelta(days=3)
	activity_id = await add_activity(
	PROBE_USER_ID, "Far Future", far.isoformat(timespec="seconds"),
	remind_mins=999_999_999,
	)
	try:
	due_ids = {row["id"] for row in await get_due_activity_reminders()}
	self.assertNotIn(
	activity_id, due_ids,
	"an absurd remind_mins must not fire a far-future activity",
	)
	finally:
	await delete_activity(activity_id, PROBE_USER_ID)

	async def _check_question_mark_translation(self, db) -> None:
	await db.execute(
	"INSERT INTO users (user_id, name, username) VALUES (?, ?, ?)",
	(PROBE_USER_ID, "Question Mark", "qm"),
	)
	async with db.execute(
	"SELECT name FROM users WHERE user_id = ?", (PROBE_USER_ID,)
	) as cursor:
	row = await cursor.fetchone()
	self.assertEqual(row[0], "Question Mark")

	async def _check_literal_question_mark(self, db) -> None:
	await db.execute(
	"INSERT INTO users (user_id, name, username) VALUES (?, ?, ?)",
	(PROBE_USER_ID, "sudah makan?", "qmlit"),
	)
	value = await db.execute(
	"SELECT name FROM users WHERE user_id = ?", (PROBE_USER_ID,)
	).fetchval()
	self.assertEqual(value, "sudah makan?")

	async def _check_transaction_rollback(self, db) -> None:
	async def _fail(tx) -> None:
	await tx.execute(
	"INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
	(PROBE_USER_ID, "Tx Rollback", "txrb"),
	)
	# Second insert violates the PRIMARY KEY on user_id.
	await tx.execute(
	"INSERT INTO users (user_id, name, username) VALUES ($1, $2, $3)",
	(PROBE_USER_ID, "Duplicate", "txdup"),
	)

	with self.assertRaises(Exception):
	async with db.transaction() as tx:
	await _fail(tx)

	exists = await db.execute(
	"SELECT 1 FROM users WHERE user_id = $1", (PROBE_USER_ID,)
	).fetchone()
	self.assertIsNone(exists, "transaction was not rolled back")

	# ── Suite ────────────────────────────────────────────────────────────────

	def test_all_checks_against_real_database(self) -> None:
	asyncio.run(self._run_all_checks())

	async def _run_all_checks(self) -> None:
	from database.connection import init_db

	db = await init_db()

	checks = [
	("migrations idempotent", self._check_migrations_idempotent),
	("expected tables exist", self._check_tables),
	("await db.execute pattern", self._check_await_execute),
	("async with cursor pattern", self._check_async_with_cursor),
	("RETURNING id + fetchval", self._check_returning_id),
	("meal_type is populated", self._check_meal_type_inference),
	("today follows user timezone", self._check_today_follows_user_timezone),
	("? placeholder translation", self._check_question_mark_translation),
	("literal ? preserved", self._check_literal_question_mark),
	("transaction rollback", self._check_transaction_rollback),
	("timestamptz due reminders", self._check_activity_due_reminder),
	("remind_mins bounded", self._check_remind_mins_is_bounded),
	]

	failures = []
	for name, check in checks:
	await self._reset_probe(db)
	try:
	await check(db)
	print(f" PASS {name}")
	except Exception as exc: # noqa: BLE001
	print(f" FAIL {name}: {exc}")
	failures.append((name, exc))

	# Always clean the probe user so the next run starts clean.
	await self._reset_probe(db)

	await self._reset_probe(db)
	# Close in the same event loop that created the pool.
	from database.connection import close_db
	await close_db()

	if failures:
	self.fail(f"{len(failures)} check(s) failed: {[n for n, _ in failures]}")


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