FabledCurator/tests/test_translate_posts.py

"""translate_posts sweep (#143) — integration, with a stubbed Interpreter client
and the task's session factory pointed at the test's db_sync."""
import pytest
from sqlalchemy import select

from backend.app.models import Artist, ImportSettings, Post
from backend.app.services import interpreter_client as ic
from backend.app.tasks.translation import retranslate_posts, translate_posts

pytestmark = pytest.mark.integration


class _Ctx:
    def __init__(self, s):
        self.s = s

    def __enter__(self):
        return self.s

    def __exit__(self, *a):
        return False


def _sf(db_sync):
    class _SM:
        def __call__(self):
            return _Ctx(db_sync)

    return _SM()


def _artist(db, name="A", slug="a"):
    a = Artist(name=name, slug=slug)
    db.add(a)
    db.flush()
    return a


def _post(db, artist_id, *, title=None, desc=None, ext="p1"):
    p = Post(
        artist_id=artist_id, external_post_id=ext,
        post_title=title, description=desc,
    )
    db.add(p)
    db.flush()
    return p


def _mark_translated(p, *, lang="ja", ver="v1"):
    """Simulate a prior (old-model) translation so retranslate has something to
    reset."""
    p.post_title_translated = "OLD"
    p.description_translated = "OLD"
    p.translated_source_lang = lang
    p.translation_engine_version = ver


def _enable(db, url="http://i.lan"):
    cfg = db.execute(
        select(ImportSettings).where(ImportSettings.id == 1)
    ).scalar_one()
    cfg.translation_enabled = True
    cfg.interpreter_base_url = url
    db.flush()


def _patch(monkeypatch, db_sync):
    monkeypatch.setattr(
        "backend.app.tasks.translation._sync_session_factory",
        lambda: _sf(db_sync),
    )


def test_translate_posts_stores_translation(db_sync, monkeypatch):
    _patch(monkeypatch, db_sync)
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
    monkeypatch.setattr(
        "backend.app.tasks.translation.ic.translate",
        lambda texts, **k: {
            "translations": [f"EN:{t}" for t in texts],
            "detected_lang": "ja", "engine": "llm", "engine_version": "v1",
        },
    )
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", desc="かわいい")
    _enable(db_sync)
    db_sync.commit()

    assert "translated=1" in translate_posts()
    db_sync.refresh(p)
    assert p.post_title_translated == "EN:ねこ"
    assert p.description_translated == "EN:かわいい"
    assert p.translated_source_lang == "ja"
    assert p.translation_engine_version == "v1"


def test_translate_posts_passthrough_english_marks_handled(db_sync, monkeypatch):
    _patch(monkeypatch, db_sync)
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
    monkeypatch.setattr(
        "backend.app.tasks.translation.ic.translate",
        lambda texts, **k: {
            "translations": list(texts), "detected_lang": "en",
            "engine": "none", "engine_version": None,
        },
    )
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="hello world", ext="p2")
    _enable(db_sync)
    db_sync.commit()

    translate_posts()
    db_sync.refresh(p)
    assert p.translated_source_lang == "en"     # marked handled...
    assert p.post_title_translated is None       # ...but nothing to show


def test_translate_posts_disabled_is_noop(db_sync, monkeypatch):
    _patch(monkeypatch, db_sync)
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", ext="p3")
    db_sync.commit()   # translation_enabled defaults False
    assert translate_posts() == "disabled"
    db_sync.refresh(p)
    assert p.translated_source_lang is None


def test_translate_posts_service_down_leaves_untranslated(db_sync, monkeypatch):
    _patch(monkeypatch, db_sync)
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: False)
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", ext="p4")
    _enable(db_sync)
    db_sync.commit()
    assert translate_posts() == "interpreter unavailable"
    db_sync.refresh(p)
    assert p.translated_source_lang is None       # will retry next run


def test_translate_posts_interrupt_reenqueues_after_backoff(db_sync, monkeypatch):
    # A drain mid-sweep (health passed, translate 503s w/ Retry-After) re-enqueues
    # the daily sweep after the backoff instead of waiting for tomorrow's beat.
    _patch(monkeypatch, db_sync)
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)

    def _drain(texts, **k):
        raise ic.InterpreterUnavailable("draining", retry_after=30)

    monkeypatch.setattr("backend.app.tasks.translation.ic.translate", _drain)
    calls = []
    monkeypatch.setattr(
        "backend.app.tasks.translation.translate_posts.apply_async",
        lambda *a, **k: calls.append((a, k)),
    )
    a = _artist(db_sync)
    _post(db_sync, a.id, title="ねこ", ext="p1")
    _enable(db_sync)
    db_sync.commit()

    assert "interrupted" in translate_posts()
    assert len(calls) == 1
    assert calls[0][1]["countdown"] == 30


def _mock_new_model(monkeypatch, *, ver="v2"):
    """Interpreter is up and translates with a NEW engine version."""
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)
    monkeypatch.setattr(
        "backend.app.tasks.translation.ic.translate",
        lambda texts, **k: {
            "translations": [f"NEW:{t}" for t in texts],
            "detected_lang": "ja", "engine": "llm", "engine_version": ver,
        },
    )


def test_retranslate_resets_and_reruns(db_sync, monkeypatch):
    # A post translated by the old model gets cleared + re-run through the new
    # one (new engine_version proves the reset happened, not a cache short-circuit).
    _patch(monkeypatch, db_sync)
    _mock_new_model(monkeypatch)
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", desc="かわいい")
    _mark_translated(p)
    _enable(db_sync)
    db_sync.commit()

    retranslate_posts(artist_ids=[a.id])
    db_sync.refresh(p)
    assert p.post_title_translated == "NEW:ねこ"
    assert p.description_translated == "NEW:かわいい"
    assert p.translation_engine_version == "v2"


def test_retranslate_artist_scope_leaves_others(db_sync, monkeypatch):
    # Scoping to one artist must not touch another artist's translations.
    _patch(monkeypatch, db_sync)
    _mock_new_model(monkeypatch)
    a1 = _artist(db_sync)
    a2 = _artist(db_sync, name="B", slug="b")
    p1 = _post(db_sync, a1.id, title="ねこ", ext="p1")
    p2 = _post(db_sync, a2.id, title="いぬ", ext="p2")
    _mark_translated(p1)
    _mark_translated(p2)
    _enable(db_sync)
    db_sync.commit()

    retranslate_posts(artist_ids=[a1.id])
    db_sync.refresh(p1)
    db_sync.refresh(p2)
    assert p1.post_title_translated == "NEW:ねこ"     # re-run
    assert p1.translation_engine_version == "v2"
    assert p2.post_title_translated == "OLD"          # untouched
    assert p2.translation_engine_version == "v1"


def test_retranslate_runs_until_done(db_sync, monkeypatch):
    # With the chunk capped at 1 and 2 posts to redo, the first chunk re-enqueues
    # itself (with _reset_done=True so it doesn't wipe the fresh work) to finish.
    _patch(monkeypatch, db_sync)
    _mock_new_model(monkeypatch)
    monkeypatch.setattr("backend.app.tasks.translation._MAX_POSTS_PER_RUN", 1)
    calls = []
    monkeypatch.setattr(
        "backend.app.tasks.translation.retranslate_posts.apply_async",
        lambda *a, **k: calls.append((a, k)),
    )
    a = _artist(db_sync)
    p1 = _post(db_sync, a.id, title="ねこ", ext="p1")
    p2 = _post(db_sync, a.id, title="いぬ", ext="p2")
    _mark_translated(p1)
    _mark_translated(p2)
    _enable(db_sync)
    db_sync.commit()

    retranslate_posts(artist_ids=[a.id])
    assert len(calls) == 1                             # one re-enqueue for the tail
    args, _kw = calls[0]
    assert args[1]["_reset_done"] is True
    assert args[1]["artist_ids"] == [a.id]


def test_retranslate_disabled_does_not_reset(db_sync, monkeypatch):
    # Never wipe translations we can't rebuild — a disabled service leaves them.
    _patch(monkeypatch, db_sync)
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", ext="p1")
    _mark_translated(p)
    db_sync.commit()   # translation_enabled defaults False
    assert retranslate_posts(artist_ids=[a.id]) == "disabled"
    db_sync.refresh(p)
    assert p.translated_source_lang == "ja"
    assert p.post_title_translated == "OLD"


def test_retranslate_unhealthy_does_not_reset(db_sync, monkeypatch):
    _patch(monkeypatch, db_sync)
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: False)
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", ext="p1")
    _mark_translated(p)
    _enable(db_sync)
    db_sync.commit()
    assert retranslate_posts(artist_ids=[a.id]) == "interpreter unavailable"
    db_sync.refresh(p)
    assert p.translated_source_lang == "ja"
    assert p.post_title_translated == "OLD"


def test_retranslate_interrupt_resumes_after_retry_after(db_sync, monkeypatch):
    # Interpreter drains mid-chunk (health passed, then translate 503s with a
    # Retry-After): the bulk re-translate re-enqueues itself after the hinted
    # backoff, with _reset_done=True so it never re-wipes fresh work — instead of
    # stalling until the daily beat.
    _patch(monkeypatch, db_sync)
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)

    def _drain(texts, **k):
        raise ic.InterpreterUnavailable("draining", retry_after=42)

    monkeypatch.setattr("backend.app.tasks.translation.ic.translate", _drain)
    calls = []
    monkeypatch.setattr(
        "backend.app.tasks.translation.retranslate_posts.apply_async",
        lambda *a, **k: calls.append((a, k)),
    )
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", ext="p1")
    _mark_translated(p)
    _enable(db_sync)
    db_sync.commit()

    assert "interrupted" in retranslate_posts(artist_ids=[a.id])
    assert len(calls) == 1                             # resume re-enqueued
    args, kw = calls[0]
    assert args[1]["_reset_done"] is True
    assert args[1]["artist_ids"] == [a.id]
    assert kw["countdown"] == 42                       # honoured the Retry-After
    db_sync.refresh(p)
    assert p.translated_source_lang is None            # reset happened, not re-run


def test_retranslate_interrupt_default_backoff_without_hint(db_sync, monkeypatch):
    # No Retry-After header → fall back to the default backoff (60s).
    _patch(monkeypatch, db_sync)
    monkeypatch.setattr("backend.app.tasks.translation.ic.health", lambda *a, **k: True)

    def _drain(texts, **k):
        raise ic.InterpreterUnavailable("draining")

    monkeypatch.setattr("backend.app.tasks.translation.ic.translate", _drain)
    calls = []
    monkeypatch.setattr(
        "backend.app.tasks.translation.retranslate_posts.apply_async",
        lambda *a, **k: calls.append((a, k)),
    )
    a = _artist(db_sync)
    p = _post(db_sync, a.id, title="ねこ", ext="p1")
    _mark_translated(p)
    _enable(db_sync)
    db_sync.commit()

    retranslate_posts(artist_ids=[a.id])
    assert len(calls) == 1
    assert calls[0][1]["countdown"] == 60