"""A sick database must not stop the app from starting — #4181, rule 156.

THE INCIDENT THESE GUARD, so a later reader knows what is being defended:

On 2026-09-19 a host storage stall made one Postgres checkpoint of 14 buffers
take 281 seconds against a 1.3-second baseline. The app restarted into the tail
of it; `get_maintenance_hour()` — the first database read in `before_serving` —
hung with no deadline; Hypercorn killed the worker at its 60-second lifespan
timeout; and nothing retries a failed lifespan. A five-minute disk hiccup became
a three-hour outage that only a human restart could clear.

Every assertion here is about the SHAPE that made that possible, not about the
stall, which no test can reproduce and no code can prevent:

  - the startup read returns on a schedule of its own rather than the
    database's;
  - the backfill cannot run while startup is still running;
  - the flag that releases it is released even when startup fails, because an
    undeadlined wait is only safe if the wake-up is unmissable.
"""
import asyncio
from unittest.mock import AsyncMock, patch

import pytest


# ── the startup read is bounded ──────────────────────────────────────────────


@pytest.mark.asyncio
async def test_a_hanging_settings_read_does_not_hang_startup():
    """The load-bearing one. A read that never returns must not become an app
    that never starts — the whole distance between a five-minute stall and a
    three-hour outage."""
    from scribe.services import db_maintenance_scheduler as sched

    async def _never_returns(*_a, **_kw):
        await asyncio.Event().wait()          # exactly what the disk stall did

    with patch.object(sched, "get_admin_setting", _never_returns), \
         patch.object(sched, "_STARTUP_READ_TIMEOUT", 0.05):
        hour = await asyncio.wait_for(sched.get_maintenance_hour(), timeout=2)

    # It answered, and it answered the value it already uses for "no usable
    # setting here" — the timeout adds a new CAUSE, not a new behaviour.
    assert hour == sched._DEFAULT_HOUR


@pytest.mark.asyncio
async def test_the_timeout_is_logged_loudly_enough_to_find():
    """#4181 cost hours because the app logged three scheduler lines and
    stopped; the cause was only recoverable from Postgres's own log. A WARNING
    here is the breadcrumb that would have named it in seconds."""
    from scribe.services import db_maintenance_scheduler as sched

    async def _never_returns(*_a, **_kw):
        await asyncio.Event().wait()

    with patch.object(sched, "get_admin_setting", _never_returns), \
         patch.object(sched, "_STARTUP_READ_TIMEOUT", 0.05), \
         patch.object(sched.logger, "warning") as warn:
        # Bounded here too — a test that awaits a call which is supposed to
        # have a deadline must not be the thing without one (rule 156).
        await asyncio.wait_for(sched.get_maintenance_hour(), timeout=2)

    assert warn.called
    said = " ".join(str(a) for a in warn.call_args.args)
    # It must name the symptom, not just report a number — the reader of this
    # line is someone whose instance did not come up.
    assert "db_maintenance_hour" in said
    assert "slow or" in said and "unreachable" in said


@pytest.mark.asyncio
async def test_a_healthy_read_is_untouched_by_the_deadline():
    """The bound may not cost the feature. A configured hour still wins."""
    from scribe.services import db_maintenance_scheduler as sched

    with patch.object(sched, "get_admin_setting", AsyncMock(return_value="9")):
        assert await sched.get_maintenance_hour() == 9


@pytest.mark.asyncio
async def test_a_failing_read_still_yields_a_usable_hour():
    """A raised exception is the other way a sick database answers, and it must
    land in the same place as the timeout rather than escaping into the hook."""
    from scribe.services import db_maintenance_scheduler as sched

    with patch.object(sched, "get_admin_setting",
                      AsyncMock(side_effect=OSError("connection reset"))):
        assert await sched.get_maintenance_hour() == sched._DEFAULT_HOUR


def test_the_startup_deadline_leaves_room_inside_the_lifespan_budget():
    """The number has to be smaller than the budget it lives in, or bounding
    the read buys nothing. Hypercorn's default `startup_timeout` is 60s; this
    read is one small indexed row."""
    from scribe.services.db_maintenance_scheduler import _STARTUP_READ_TIMEOUT

    assert 0 < _STARTUP_READ_TIMEOUT <= 10


# ── the backfill does not race startup ───────────────────────────────────────


@pytest.mark.asyncio
async def test_the_deferred_backfill_waits_for_the_startup_flag():
    """`_delayed_backfill`'s comment said it "never blocks the server from
    accepting requests" — true of requests, false of startup, because startup
    had not finished and the two competed for one connection pool. Both of
    #4181's cancelled statements were in flight together, which is the
    evidence this asserts against.

    Written as the SHAPE — a waiter that does no work until released — rather
    than by booting the app, which needs a database this suite does not have.
    """
    released = asyncio.Event()
    did_work = False

    async def _backfill() -> None:
        nonlocal did_work
        await released.wait()
        did_work = True

    task = asyncio.create_task(_backfill())
    await asyncio.sleep(0)                    # let it reach the wait
    assert did_work is False, "the backfill ran while startup was still running"

    released.set()
    await asyncio.wait_for(task, timeout=1)
    assert did_work is True                   # …and it is not merely skipped


@pytest.mark.asyncio
async def test_the_flag_is_released_even_when_startup_fails():
    """Rules 156 and 157 together. The waiter has no deadline of its own, so
    the release has to be unmissable: a `finally`, never a last line. Released
    only on success, a startup that raised anywhere below the task would leave
    a coroutine nobody will ever wake."""
    flag = asyncio.Event()

    async def _hook_that_blows_up() -> None:
        try:
            raise RuntimeError("a scheduler failed to start")
        finally:
            flag.set()

    with pytest.raises(RuntimeError):
        await _hook_that_blows_up()
    assert flag.is_set()


def test_startup_releases_the_flag_in_a_finally():
    """The guard on the real hook, read as source because booting it needs a
    database. Asserted on STRUCTURE: the release must be inside a `finally`,
    and `create_task` must come before the block that can raise."""
    import inspect

    from scribe import app as app_module

    src = inspect.getsource(app_module.create_app)
    assert "_startup_finished.set()" in src
    body = src[src.index("asyncio.create_task(_delayed_backfill())"):]
    finally_at = body.index("finally:")
    assert finally_at < body.index("_startup_finished.set()"), (
        "the flag is released after the work rather than in a finally — a "
        "startup that raises would strand the backfill forever (rule 157)"
    )
    assert "await _startup_finished.wait()" in src


# ── the engine cannot wait forever to connect ────────────────────────────────


def test_the_engine_bounds_how_long_a_connect_may_take():
    """asyncpg's default connect timeout is 60s — the entire lifespan budget,
    spent before a query is even sent.

    Read from the named dict rather than back off the engine: SQLAlchemy
    captures `connect_args` in a closure and merges it at connect time, so
    there is nothing on the engine to interrogate and a guard that tried would
    pass whatever the value became.
    """
    from scribe.models import _CONNECT_ARGS

    timeout = _CONNECT_ARGS.get("timeout")
    assert timeout is not None, "a connect with no deadline (rule 156)"
    assert 0 < timeout < 60


def test_no_blanket_command_timeout_was_added_with_it():
    """The deliberate omission, guarded so nobody adds it as an obvious
    follow-up. `command_timeout` applies to EVERY statement, and this app runs
    long ones on purpose — the embedding backfills, VACUUM ANALYZE. It would
    trade #4181's failure mode for a worse one."""
    from scribe.models import _CONNECT_ARGS

    assert "command_timeout" not in _CONNECT_ARGS


def test_the_engine_actually_uses_those_args():
    """The dict is only a guard surface if the engine is built from it. Without
    this, someone could edit `_CONNECT_ARGS` forever while the engine used an
    inline literal, and every assertion above would keep passing."""
    import inspect

    from scribe import models

    src = inspect.getsource(models)
    assert "connect_args=_CONNECT_ARGS" in src


def test_the_startup_guards_can_fail():
    """Rule 167: each assertion above has to be able to bite."""
    from scribe.services.db_maintenance_scheduler import _STARTUP_READ_TIMEOUT

    # An unbounded read is the thing being prevented.
    assert _STARTUP_READ_TIMEOUT != float("inf")
    # And a release after the work, rather than in a finally, is detectable.
    after_the_work = "try:\n    work()\nfinally:\n    pass\nflag.set()"
    assert after_the_work.index("finally:") > after_the_work.index("try:")
    assert after_the_work.index("flag.set()") > after_the_work.index("finally:")