tests/test_modal_client.py

from backend.modal_client import ModalSynthesisClient
import pytest
import requests


def test_modal_client_warns_when_configured_with_dashboard_url() -> None:
    client = ModalSynthesisClient(base_url="https://modal.com/apps/mattkevan/main/deployed/scriptorium-tts")

    warning = client.configuration_warning()

    assert warning is not None
    assert "modal.run" in warning
    assert "dashboard" in warning.lower()


def test_modal_client_accepts_modal_run_base_url() -> None:
    client = ModalSynthesisClient(base_url="https://scriptorium-tts--mattkevan.modal.run")

    assert client.configuration_warning() is None


def test_modal_client_defaults_to_longer_timeout() -> None:
    client = ModalSynthesisClient(base_url="https://scriptorium-tts--mattkevan.modal.run")

    assert client.timeout_seconds == 300.0


def test_modal_client_uses_longer_default_timeout_from_env(monkeypatch) -> None:
    monkeypatch.delenv("SCRIPTORIUM_MODAL_TIMEOUT_SECONDS", raising=False)
    monkeypatch.delenv("SCRIPTORIUM_MODAL_POLL_INTERVAL_SECONDS", raising=False)
    monkeypatch.delenv("SCRIPTORIUM_MODAL_STATUS_RETRIES", raising=False)

    client = ModalSynthesisClient.from_env()

    assert client.timeout_seconds == 300.0
    assert client.poll_interval_seconds == 3.0
    assert client.max_status_retries == 4


def test_modal_client_surfaces_read_timeout_as_user_facing_error(monkeypatch, tmp_path) -> None:
    client = ModalSynthesisClient(base_url="https://scriptorium-tts--mattkevan.modal.run", timeout_seconds=1)

    def fake_post(*args, **kwargs):
        raise requests.ReadTimeout("timed out")

    monkeypatch.setattr("backend.modal_client.requests.post", fake_post)

    with pytest.raises(ValueError, match="Modal request timed out"):
        client.generate_preview(
            text="Hello",
            output_path=tmp_path / "preview.wav",
            voice_config=type("Voice", (), {"to_dict": lambda self: {}, "model": "omnivoice"})(),
            diffusion_steps=32,
            speed=1.0,
        )


def test_modal_client_retries_transient_status_failures(monkeypatch, tmp_path) -> None:
    client = ModalSynthesisClient(
        base_url="https://scriptorium-tts--mattkevan.modal.run",
        poll_interval_seconds=0,
        max_status_retries=2,
    )

    calls = {"count": 0}

    def fake_get_json(path, *, params):
        calls["count"] += 1
        if calls["count"] == 1:
            raise ValueError("Modal request failed: 500 Server Error")
        return {"status": "completed", "events": []}

    monkeypatch.setattr(client, "_get_json", fake_get_json)
    monkeypatch.setattr("backend.modal_client.time.sleep", lambda *_args, **_kwargs: None)

    events = list(
        client.render(
            session_id="session-a",
            job_id="job-123",
            render_dir=tmp_path,
            voice_config=type("Voice", (), {"model": "omnivoice"})(),
        )
    )

    assert calls["count"] == 2
    assert events[0]["type"] == "log"
    assert "Transient Modal status error" in events[0]["message"]