Zum Inhalt

tests/test_pause_after_transcribe.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""Tests for pause-after-transcribe behavior in /api/runs."""

import pytest
from fastapi.testclient import TestClient


@pytest.fixture
def client():
    from webgui.app import app

    return TestClient(app)


def test_run_request_accepts_pause_field(client, fixtures_dir, monkeypatch):
    """POST /api/runs with pause_after_transcribe=true should not 422."""
    # Block actual subprocess spawn — we only care about request validation
    from webgui import app as app_mod

    calls = []

    def fake_spawn_pipeline(**kwargs):
        calls.append(kwargs)

        class FakeJob:
            pass

        return FakeJob()

    monkeypatch.setattr(app_mod, "spawn_pipeline", fake_spawn_pipeline)
    monkeypatch.setattr(app_mod.registry, "_slot", None)

    r = client.post(
        "/api/runs",
        json={
            "audio": str(fixtures_dir / "sample.m4a"),
            "viz": "dialogue",
            "language": "auto",
            "model": "tiny",
            "diarize": "off",
            "episode": "EP TEST",
            "show_name": "Signal",
            "skip_transcribe": False,
            "skip_meta": False,
            "skip_render": False,
            "skip_upload": True,
            "pause_after_transcribe": True,
        },
        follow_redirects=False,
    )
    assert r.status_code in (303, 200), r.text
    # spawn_pipeline was called
    assert len(calls) == 1


def test_pause_flag_forces_skip_meta_render_upload(client, fixtures_dir, monkeypatch):
    """When pause=true, the spawned command must include skip flags."""
    from webgui import app as app_mod

    captured_cmd = []

    def fake_spawn_pipeline(cmd, **kwargs):
        captured_cmd.append(cmd)

        class FakeJob:
            pass

        return FakeJob()

    monkeypatch.setattr(app_mod, "spawn_pipeline", fake_spawn_pipeline)
    monkeypatch.setattr(app_mod.registry, "_slot", None)

    client.post(
        "/api/runs",
        json={
            "audio": str(fixtures_dir / "sample.m4a"),
            "viz": "dialogue",
            "language": "auto",
            "model": "tiny",
            "diarize": "off",
            "episode": "EP TEST",
            "show_name": "Signal",
            "skip_transcribe": False,
            "skip_meta": False,
            "skip_render": False,
            "skip_upload": False,
            "pause_after_transcribe": True,
        },
        follow_redirects=False,
    )
    assert len(captured_cmd) == 1
    cmd = captured_cmd[0]
    cmd_str = " ".join(cmd)
    assert "--skip-meta" in cmd_str
    assert "--skip-render" in cmd_str
    assert "--skip-upload" in cmd_str


def test_pause_flag_default_false_does_not_inject_skips(client, fixtures_dir, monkeypatch):
    """Without pause flag, skip flags follow the explicit request fields only."""
    from webgui import app as app_mod

    captured_cmd = []

    def fake_spawn_pipeline(cmd, **kwargs):
        captured_cmd.append(cmd)

        class FakeJob:
            pass

        return FakeJob()

    monkeypatch.setattr(app_mod, "spawn_pipeline", fake_spawn_pipeline)
    monkeypatch.setattr(app_mod.registry, "_slot", None)

    client.post(
        "/api/runs",
        json={
            "audio": str(fixtures_dir / "sample.m4a"),
            "viz": "dialogue",
            "language": "auto",
            "model": "tiny",
            "diarize": "off",
            "episode": "EP TEST",
            "show_name": "Signal",
            "skip_transcribe": False,
            "skip_meta": False,
            "skip_render": False,
            "skip_upload": False,
            "pause_after_transcribe": False,
        },
        follow_redirects=False,
    )
    cmd_str = " ".join(captured_cmd[0])
    assert "--skip-meta" not in cmd_str
    assert "--skip-render" not in cmd_str
    # skip_upload was False → no flag injected
    assert "--skip-upload" not in cmd_str