FabledCurator/tests/test_scheduler_service.py

"""FC-3d: scheduler_service unit tests.

Covers compute_effective_interval (override/artist/default precedence,
backoff exponent at failure counts 0/1/3/6/10, floor and ceiling clamp)
and select_due_sources (auto_check off, never-checked, enabled=False,
due/not-due boundary).
"""
from datetime import UTC, datetime, timedelta

import pytest

from backend.app.models import Artist, ImportSettings, Source
from backend.app.services.scheduler_service import (
    backfill_ready,
    compute_effective_interval,
    scheduler_status,
    select_due_sources,
)

pytestmark = pytest.mark.integration


# --- compute_effective_interval (pure, no DB) -----------------------------


def _src(override=None, failures=0):
    return Source(
        artist_id=1, platform="patreon", url="https://x", enabled=True,
        check_interval_override=override, consecutive_failures=failures,
    )


def _art(interval=None, auto=True):
    return Artist(name="A", slug="a", auto_check=auto, check_interval_seconds=interval)


def _settings(default=28800):
    return ImportSettings(id=1, download_schedule_default_seconds=default)


def test_interval_uses_source_override_when_set():
    assert compute_effective_interval(_src(override=600), _art(interval=3600), _settings()) == 600


def test_interval_falls_back_to_artist_interval():
    assert compute_effective_interval(_src(), _art(interval=3600), _settings()) == 3600


def test_interval_falls_back_to_global_default():
    assert compute_effective_interval(_src(), _art(), _settings(default=900)) == 900


def test_interval_backoff_zero_failures_is_identity():
    assert compute_effective_interval(_src(override=600), _art(), _settings()) == 600


def test_interval_backoff_one_failure_doubles():
    assert compute_effective_interval(_src(override=600, failures=1), _art(), _settings()) == 1200


def test_interval_backoff_three_failures_x8():
    assert compute_effective_interval(_src(override=600, failures=3), _art(), _settings()) == 4800


def test_interval_backoff_caps_at_exponent_six():
    # 600 * 2^6 = 38400; 2^7 would be 76800 but exponent caps at 6.
    assert compute_effective_interval(_src(override=600, failures=6), _art(), _settings()) == 38400
    assert compute_effective_interval(_src(override=600, failures=10), _art(), _settings()) == 38400


def test_interval_floor_at_60_seconds():
    # base 30 with 0 failures would be 30s; floor lifts to 60.
    assert compute_effective_interval(_src(override=30), _art(), _settings()) == 60


def test_interval_ceiling_at_86400_seconds():
    # base 28800 * 2^6 = 1843200; ceiling caps at 86400.
    assert compute_effective_interval(_src(failures=6), _art(), _settings(default=28800)) == 86400


# --- select_due_sources (real session) ------------------------------------


async def _seed_artist(db, *, auto=True, interval=None, name="A"):
    a = Artist(name=name, slug=name.lower(), auto_check=auto, check_interval_seconds=interval)
    db.add(a)
    await db.flush()
    return a


@pytest.mark.asyncio
async def test_select_skips_disabled_sources(db):
    artist = await _seed_artist(db, name="sel-disabled")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://sel-disabled",
        enabled=False, consecutive_failures=0,
    ))
    await db.commit()
    due = await select_due_sources(db)
    assert all(s.url != "https://sel-disabled" for s in due)


@pytest.mark.asyncio
async def test_select_skips_artist_auto_check_false(db):
    artist = await _seed_artist(db, auto=False, name="sel-noauto")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://sel-noauto",
        enabled=True, consecutive_failures=0,
    ))
    await db.commit()
    due = await select_due_sources(db)
    assert all(s.url != "https://sel-noauto" for s in due)


@pytest.mark.asyncio
async def test_select_includes_never_checked(db):
    artist = await _seed_artist(db, name="sel-never")
    src = Source(
        artist_id=artist.id, platform="patreon", url="https://sel-never",
        enabled=True, consecutive_failures=0,
    )
    db.add(src)
    await db.commit()
    due = await select_due_sources(db)
    assert any(s.url == "https://sel-never" for s in due)


@pytest.mark.asyncio
async def test_select_excludes_not_yet_due(db):
    artist = await _seed_artist(db, interval=3600, name="sel-recent")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://sel-recent",
        enabled=True, consecutive_failures=0,
        last_checked_at=datetime.now(UTC) - timedelta(seconds=60),
    ))
    await db.commit()
    due = await select_due_sources(db)
    assert all(s.url != "https://sel-recent" for s in due)


@pytest.mark.asyncio
async def test_select_includes_past_due(db):
    artist = await _seed_artist(db, interval=60, name="sel-past")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://sel-past",
        enabled=True, consecutive_failures=0,
        last_checked_at=datetime.now(UTC) - timedelta(seconds=3600),
    ))
    await db.commit()
    due = await select_due_sources(db)
    assert any(s.url == "https://sel-past" for s in due)


# --- a running backfill is due every tick (2026-09-25) ---------------------


def _backfilling(runs=5, failures=0, state="running"):
    src = _src(failures=failures)
    src.config_overrides = {"_backfill_state": state}
    src.backfill_runs_remaining = runs
    return src


def test_a_running_backfill_is_ready():
    assert backfill_ready(_backfilling()) is True


def test_a_backfill_out_of_budget_or_not_running_is_not_ready():
    assert backfill_ready(_backfilling(runs=0)) is False
    assert backfill_ready(_backfilling(state="stalled")) is False
    assert backfill_ready(_src()) is False


def test_a_failing_backfill_falls_back_to_its_backoff():
    """Otherwise a source whose chunks keep erroring would retry every minute."""
    assert backfill_ready(_backfilling(failures=1)) is False


@pytest.mark.asyncio
async def test_select_runs_a_backfill_now_not_at_its_next_check(db):
    """Checked a minute ago on an hourly interval: not due, unless backfilling —
    the next chunk should not wait an interval (the 2026-09-25 armed-and-idle
    backfill)."""
    artist = await _seed_artist(db, interval=3600, name="bf-now")
    for url, overrides, runs in (
        ("https://bf-running", {"_backfill_state": "running"}, 5),
        ("https://bf-idle", {}, 0),
    ):
        db.add(Source(
            artist_id=artist.id, platform="patreon", url=url, enabled=True,
            consecutive_failures=0, config_overrides=overrides,
            backfill_runs_remaining=runs,
            last_checked_at=datetime.now(UTC) - timedelta(seconds=60),
        ))
    await db.commit()
    urls = {s.url for s in await select_due_sources(db)}
    assert "https://bf-running" in urls
    assert "https://bf-idle" not in urls


@pytest.mark.asyncio
async def test_a_backfill_runs_even_when_its_artist_is_not_auto_checked(db):
    """The operator started it by hand; auto-check governs the schedule only."""
    artist = await _seed_artist(db, auto=False, name="bf-noauto")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://bf-noauto",
        enabled=True, consecutive_failures=0,
        config_overrides={"_backfill_state": "running"}, backfill_runs_remaining=5,
        last_checked_at=datetime.now(UTC),
    ))
    await db.commit()
    assert any(s.url == "https://bf-noauto" for s in await select_due_sources(db))


# --- platform-rate-limit cooldown -----------------------------------------


@pytest.mark.asyncio
async def test_select_skips_sources_on_platform_in_cooldown(db):
    """After set_platform_cooldown('patreon'), select_due_sources excludes
    every patreon source — even ones that would otherwise be past-due."""
    from backend.app.services.scheduler_service import set_platform_cooldown

    artist = await _seed_artist(db, interval=60, name="cooldown-p")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://cd-p",
        enabled=True, consecutive_failures=0,
        last_checked_at=datetime.now(UTC) - timedelta(seconds=3600),
    ))
    await db.commit()

    # Sanity-pre: due before the cooldown is set.
    due = await select_due_sources(db)
    assert any(s.url == "https://cd-p" for s in due)

    await set_platform_cooldown(db, "patreon")
    await db.commit()

    due = await select_due_sources(db)
    assert all(s.url != "https://cd-p" for s in due)


@pytest.mark.asyncio
async def test_cooldown_is_per_platform_other_platforms_unaffected(db):
    """A cooldown on patreon doesn't block subscribestar sources from
    being due — cooldowns are scoped to the platform that hit the limit."""
    from backend.app.services.scheduler_service import set_platform_cooldown

    artist = await _seed_artist(db, interval=60, name="cooldown-mixed")
    db.add_all([
        Source(
            artist_id=artist.id, platform="patreon", url="https://mix-p",
            enabled=True, consecutive_failures=0,
            last_checked_at=datetime.now(UTC) - timedelta(seconds=3600),
        ),
        Source(
            artist_id=artist.id, platform="subscribestar", url="https://mix-ss",
            enabled=True, consecutive_failures=0,
            last_checked_at=datetime.now(UTC) - timedelta(seconds=3600),
        ),
    ])
    await db.commit()

    await set_platform_cooldown(db, "patreon")
    await db.commit()

    due = await select_due_sources(db)
    urls = {s.url for s in due}
    assert "https://mix-ss" in urls
    assert "https://mix-p" not in urls


@pytest.mark.asyncio
async def test_select_ignores_expired_cooldown(db):
    """An expired cooldown row doesn't filter — routing treats expired
    same as not-set. The stale row stays around until a future sweep
    prunes it."""
    from backend.app.models import AppSetting
    from backend.app.services.scheduler_service import (
        PLATFORM_COOLDOWN_KEY_PREFIX,
    )

    artist = await _seed_artist(db, interval=60, name="cooldown-expired")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://cd-exp",
        enabled=True, consecutive_failures=0,
        last_checked_at=datetime.now(UTC) - timedelta(seconds=3600),
    ))
    db.add(AppSetting(
        key=f"{PLATFORM_COOLDOWN_KEY_PREFIX}patreon",
        value=(datetime.now(UTC) - timedelta(seconds=10)).isoformat(),
    ))
    await db.commit()

    due = await select_due_sources(db)
    assert any(s.url == "https://cd-exp" for s in due)


@pytest.mark.asyncio
async def test_select_orders_most_overdue_first_then_id(db):
    """Within the due set, sources are ordered by last_checked_at ASC NULLS
    FIRST. Combined with Celery FIFO on the download queue, the most-overdue
    source in each tick reaches a worker first — preventing the 'a freshly-
    rerun source keeps cutting in line ahead of one that's still waiting'
    starvation pattern when queue throughput is below the tick population."""
    artist = await _seed_artist(db, interval=60, name="order-test")
    now = datetime.now(UTC)
    # Newest check — least overdue (but still past its 60s interval).
    s_new = Source(
        artist_id=artist.id, platform="patreon", url="https://order-new",
        enabled=True, consecutive_failures=0,
        last_checked_at=now - timedelta(minutes=2),
    )
    # Oldest check — most overdue among checked sources.
    s_old = Source(
        artist_id=artist.id, platform="patreon", url="https://order-old",
        enabled=True, consecutive_failures=0,
        last_checked_at=now - timedelta(hours=4),
    )
    # Never checked — wins the ordering (NULLS FIRST).
    s_never = Source(
        artist_id=artist.id, platform="patreon", url="https://order-never",
        enabled=True, consecutive_failures=0,
    )
    db.add_all([s_new, s_old, s_never])
    await db.commit()

    due = await select_due_sources(db)
    # Filter to just our test seeds (concurrent tests may add others).
    urls = [s.url for s in due if s.url.startswith("https://order-")]
    assert urls == [
        "https://order-never",
        "https://order-old",
        "https://order-new",
    ]


@pytest.mark.asyncio
async def test_set_platform_cooldown_upserts(db):
    """Calling set_platform_cooldown twice on the same platform updates
    the row in place rather than inserting a duplicate (AppSetting's PK
    is `key`)."""
    from sqlalchemy import select as sa_select

    from backend.app.models import AppSetting
    from backend.app.services.scheduler_service import (
        PLATFORM_COOLDOWN_KEY_PREFIX,
        set_platform_cooldown,
    )

    await set_platform_cooldown(db, "patreon", seconds=60)
    await db.commit()
    await set_platform_cooldown(db, "patreon", seconds=600)
    await db.commit()

    rows = (await db.execute(
        sa_select(AppSetting).where(
            AppSetting.key == f"{PLATFORM_COOLDOWN_KEY_PREFIX}patreon"
        )
    )).scalars().all()
    assert len(rows) == 1


# --- scheduler_status ingestion counts (#387 B3) --------------------------


@pytest.mark.asyncio
async def test_status_counts_failing_and_no_access_separately(db):
    """The front-door ribbon's two numbers must not bleed into each other.

    A tier-limited source has consecutive_failures == 0 by construction (its
    run succeeded), so counting failures by `last_error IS NOT NULL` or by the
    presence of an error_type would report it as broken — which is the exact
    claim phase A exists to stop the app making.
    """
    artist = await _seed_artist(db, name="cnt")
    db.add_all([
        Source(artist_id=artist.id, platform="patreon", url="https://cnt-broken",
               enabled=True, consecutive_failures=3, error_type="auth_error"),
        Source(artist_id=artist.id, platform="patreon", url="https://cnt-gated",
               enabled=True, consecutive_failures=0, error_type="tier_limited"),
        Source(artist_id=artist.id, platform="patreon", url="https://cnt-fine",
               enabled=True, consecutive_failures=0),
    ])
    await db.commit()

    status = await scheduler_status(db)
    assert status["failing_sources"] == 1
    assert status["no_access_sources"] == 1


@pytest.mark.asyncio
async def test_status_counts_ignore_disabled_sources(db):
    """Disabling is the operator's way of parking a sub they stopped paying for
    (#1285 clears its state on the way). A parked source must not keep a number
    lit on the front door."""
    artist = await _seed_artist(db, name="cnt-off")
    db.add_all([
        Source(artist_id=artist.id, platform="patreon", url="https://cnt-off-broken",
               enabled=False, consecutive_failures=5),
        Source(artist_id=artist.id, platform="patreon", url="https://cnt-off-gated",
               enabled=False, error_type="tier_limited"),
    ])
    await db.commit()

    status = await scheduler_status(db)
    assert status["failing_sources"] == 0
    assert status["no_access_sources"] == 0


@pytest.mark.asyncio
async def test_status_counts_are_not_scoped_to_auto_check(db):
    """A source erroring on a manual-only artist is still erroring. The counts
    deliberately span all enabled sources, unlike `auto_sources` above which
    only describes what is on a schedule."""
    artist = await _seed_artist(db, auto=False, name="cnt-manual")
    db.add(Source(
        artist_id=artist.id, platform="patreon", url="https://cnt-manual",
        enabled=True, consecutive_failures=2,
    ))
    await db.commit()

    status = await scheduler_status(db)
    assert status["failing_sources"] == 1
    assert status["auto_sources"] == 0