serverAppConfigs/tests/frank_tts_unit.bats

#!/usr/bin/env bats
# Seam: frank-tts, the cloned-voice speech façade on Dankclaw :7772.
#
# The contract this pins is the one every consumer (omp, OpenWebUI,
# open-notebook) codes against: POST /v1/audio/speech with voice=<name> is
# answered with audio in THAT voice pack, where "that voice pack" is exactly
# $VOICES_DIR/<name>/reference.wav — the file that makes a name real. A name
# without a reference must come back as a 400 naming the voices that do exist
# (a silent fallback voice is how a wrong-voice endpoint stays wrong for
# weeks), the harness' voice_id must work as an alias for voice, oversized text
# must be refused rather than half-narrated, and the mp3 response_format must
# really be mp3. Whatever engine sits behind it is a separate decision — so
# both backends are driven here through stubs, never against a GPU.
#
# Everything runs against the real app.py over 127.0.0.1 ephemeral ports with
# a temp VOICES_DIR: no Dankclaw, no network, no containers, no pkill — the
# processes this file starts are recorded by pid and killed in teardown.

load helpers

setup() {
    APP="$REPO_ROOT/dankclaw/voiceservice/app.py"
    UNIT="$REPO_ROOT/dankclaw/voiceservice/quadlet/frank-tts.container"
    ENV_FILE="$REPO_ROOT/scripts/frank-tts.env"
    TEST_TMP="$(mktemp -d)"
    PIDS="$TEST_TMP/pids"
    : > "$PIDS"

    # One real voice pack and one directory that LOOKS like a pack but has no
    # reference.wav — it must never be offered as a voice.
    VOICES="$TEST_TMP/voices"
    mkdir -p "$VOICES/narrator1" "$VOICES/nopack"
    python3 -c 'import sys; sys.stdout.buffer.write(b"RIFF" + bytes(range(256)) * 4)' \
        > "$VOICES/narrator1/reference.wav"
    printf 'a manifest is not a voice\n' > "$VOICES/nopack/manifest.json"

    # The audio the stub upstream "synthesises": distinct bytes, so a response
    # can be proven to be the engine's output and not something the shim made
    # up on its way past.
    python3 -c 'import sys; sys.stdout.buffer.write(b"RIFF" + bytes(reversed(range(128))) * 3)' \
        > "$TEST_TMP/upstream.wav"

    write_stub
}

teardown() {
    # Kill only what this file started, by recorded pid.
    while IFS= read -r pid; do
        [ -n "$pid" ] && kill "$pid" 2>/dev/null || true
    done < "$PIDS"
    wait 2>/dev/null || true
    rm -rf "$TEST_TMP"
}

# write_stub — the upstream TTS server arm A forwards to: records the JSON body
# it was handed (so "did the reference voice actually travel?" is observable)
# and answers with $STUB_WAV. Prints its bound port, chosen by the kernel.
write_stub() {
    cat > "$TEST_TMP/stub_upstream.py" <<'PY'
import json, os
from http.server import BaseHTTPRequestHandler, HTTPServer

class Stub(BaseHTTPRequestHandler):
    def do_POST(self):
        length = int(self.headers.get('content-length') or 0)
        with open(os.environ['STUB_CAPTURE'], 'w') as fh:
            json.dump(json.loads(self.rfile.read(length) or b'{}'), fh)
        with open(os.environ['STUB_WAV'], 'rb') as fh:
            audio = fh.read()
        self.send_response(200)
        self.send_header('content-type', 'audio/wav')
        self.send_header('content-length', str(len(audio)))
        self.end_headers()
        self.wfile.write(audio)

    def log_message(self, *args):
        pass

server = HTTPServer(('127.0.0.1', 0), Stub)
print(server.server_address[1], flush=True)
server.serve_forever()
PY
}

start_stub() {
    : > "$TEST_TMP/stub.log"
    STUB_WAV="$TEST_TMP/upstream.wav" STUB_CAPTURE="$TEST_TMP/capture.json" \
        python3 "$TEST_TMP/stub_upstream.py" > "$TEST_TMP/stub.log" 2>&1 &
    echo "$!" >> "$PIDS"
    local waited=0
    while [ "$waited" -lt 100 ]; do
        STUB_PORT="$(head -n1 "$TEST_TMP/stub.log")"
        [ -n "$STUB_PORT" ] && return 0
        sleep 0.05
        waited=$((waited + 1))
    done
    echo "stub upstream never bound a port:"
    cat "$TEST_TMP/stub.log"
    return 1
}

# start_app — boot app.py on 127.0.0.1 with an ephemeral port and read that
# port back off its own startup line: no port guessing, no race. The caller
# exports the env the run needs first; app.py reads its knobs at request time.
start_app() {
    : > "$TEST_TMP/app.log"
    python3 "$APP" --host 127.0.0.1 --port 0 > "$TEST_TMP/app.log" 2>&1 &
    echo "$!" >> "$PIDS"
    local waited=0
    while [ "$waited" -lt 200 ]; do
        APP_PORT="$(sed -nE 's|.*listening on http://127\.0\.0\.1:([0-9]+)|\1|p' "$TEST_TMP/app.log" | head -n1)"
        [ -n "$APP_PORT" ] && return 0
        sleep 0.05
        waited=$((waited + 1))
    done
    echo "frank-tts never came up:"
    cat "$TEST_TMP/app.log"
    return 1
}

# arm_omni — the shim forwarding to an OpenAI-shaped upstream (arm A).
arm_omni() {
    start_stub
    export VOICES_DIR="$VOICES"
    export TTS_BACKEND=omni-http
    export TTS_UPSTREAM="http://127.0.0.1:$STUB_PORT"
}

speech() {
    # speech <json-payload-inline> — the HTTP code lands in $output, the body
    # in $TEST_TMP/out.bin and the response headers in head.txt.
    local data="$1"
    run curl -sS -o "$TEST_TMP/out.bin" -D "$TEST_TMP/head.txt" -w '%{http_code}' \
        -H 'Content-Type: application/json' --data-binary "$data" \
        "http://127.0.0.1:$APP_PORT/v1/audio/speech"
}

@test "app.py parses as python the runtime can import" {
    run env PYTHONPYCACHEPREFIX="$TEST_TMP/pycache" python3 -m py_compile "$APP"
    [ "$status" -eq 0 ] || { echo "$output"; false; }
}

@test "voice=narrator1 returns the engine's audio with the pack's reference attached" {
    arm_omni
    start_app
    speech '{"model":"tts-1","input":"hello world","voice":"narrator1"}'
    [ "$status" -eq 0 ] || { cat "$TEST_TMP/app.log"; false; }
    [ "$output" = "200" ] || { cat "$TEST_TMP/app.log"; false; }
    grep -qi '^content-type: audio/wav' "$TEST_TMP/head.txt"
    cmp "$TEST_TMP/out.bin" "$TEST_TMP/upstream.wav"
    # The voice only arrives at the engine as base64 reference_audio — prove it
    # is the pack's bytes, not a path and not the engine's own default voice.
    python3 - "$TEST_TMP/capture.json" "$VOICES/narrator1/reference.wav" <<'PY'
import base64, json, sys
body = json.load(open(sys.argv[1]))
reference = open(sys.argv[2], 'rb').read()
assert body['voice'] == 'narrator1', body
assert body['input'] == 'hello world', body
assert body['response_format'] == 'wav', body
assert base64.b64decode(body['reference_audio']) == reference, 'reference_audio is not the pack wav'
PY
    # What a consumer discovers before it speaks: /health and the voice list.
    # A directory with a manifest but no reference.wav is not a voice.
    run curl -sS "http://127.0.0.1:$APP_PORT/v1/audio/voices"
    [ "$output" = '["narrator1"]' ]
    run curl -sS "http://127.0.0.1:$APP_PORT/health"
    [[ "$output" == *'"status": "ok"'* ]]
    [[ "$output" == *'"backend": "omni-http"'* ]]
    [[ "$output" == *'"voices": ["narrator1"]'* ]]
}

@test "voice_id works as the alias the omp harness sends" {
    arm_omni
    start_app
    speech '{"model":"tts-1","input":"hello","voice_id":"narrator1"}'
    [ "$output" = "200" ] || { cat "$TEST_TMP/app.log"; false; }
    cmp "$TEST_TMP/out.bin" "$TEST_TMP/upstream.wav"
    python3 - "$TEST_TMP/capture.json" <<'PY'
import json, sys
body = json.load(open(sys.argv[1]))
assert body['voice'] == 'narrator1', 'voice_id never became the upstream voice'
assert 'voice_id' not in body, body
PY
}

@test "an unknown voice is a 400 naming the voices that exist" {
    arm_omni
    start_app
    speech '{"model":"tts-1","input":"hello","voice":"nosuch"}'
    [ "$output" = "400" ]
    grep -q 'narrator1' "$TEST_TMP/out.bin"
    # Refused before the engine saw anything: no audio, no upstream call.
    [ ! -s "$TEST_TMP/capture.json" ]
}

@test "oversized input is refused outright, not narrated half-way" {
    arm_omni
    # The limit is a knob the operator sets, not a constant in the seam: the
    # app reads its knobs at request time, so exporting it before boot works.
    export SHIM_MAX_CHARS=10
    start_app
    python3 - "$TEST_TMP/payload.json" <<'PY'
import json, sys
json.dump({'model': 'tts-1', 'input': 'A' * 11, 'voice': 'narrator1'},
          open(sys.argv[1], 'w'))
PY
    run curl -sS -o "$TEST_TMP/out.bin" -w '%{http_code}' \
        -H 'Content-Type: application/json' --data-binary "@$TEST_TMP/payload.json" \
        "http://127.0.0.1:$APP_PORT/v1/audio/speech"
    [ "$output" = "414" ]
    [ ! -s "$TEST_TMP/capture.json" ]
}

@test "every other route is a 404, GET or POST" {
    arm_omni
    start_app
    run curl -sS -o /dev/null -w '%{http_code}' "http://127.0.0.1:$APP_PORT/v1/audio/speeck"
    [ "$output" = "404" ]
    run curl -sS -o /dev/null -w '%{http_code}' -X POST \
        -H 'Content-Type: application/json' -d '{"input":"hi","voice":"narrator1"}' \
        "http://127.0.0.1:$APP_PORT/v1/speech"
    [ "$output" = "404" ]
}

@test "an unrouted POST does not poison a reused connection" {
    # The shim speaks HTTP/1.1, so a client may keep one TCP connection alive
    # across requests. A body left unread on the way to a 404 is parsed as the
    # NEXT request line — the consumer would blame the following, perfectly
    # valid call. So: 404 first, then /health on the same connection.
    arm_omni
    start_app
    python3 - "$APP_PORT" <<'PY'
import http.client, json, sys
conn = http.client.HTTPConnection("127.0.0.1", int(sys.argv[1]), timeout=15)
conn.request("POST", "/v1/audio/speeck",
             body=json.dumps({"input": "hi", "voice": "narrator1"}),
             headers={"Content-Type": "application/json"})
first = conn.getresponse()
assert first.status == 404, first.status
first.read()
conn.request("GET", "/health")
second = conn.getresponse()
assert second.status == 200, second.status
assert json.loads(second.read())["status"] == "ok"
conn.close()
PY
}

@test "the CLI backend gets the resolved reference as --tts-speaker-file" {
    # A fake binary standing in for llama-tts: it records its argv, writes a
    # WAV where --output points, and reports the temp dir it was asked to work
    # in so the cleanup promise is observable too.
    cat > "$TEST_TMP/fake-tts" <<'FAKE'
#!/usr/bin/env bash
printf '%s\n' "$@" > "$ARGS_FILE"
prev=""
for arg in "$@"; do
    if [ "$prev" = "--output" ]; then
        printf 'CLI-AUDIO' > "$arg"
        dirname "$arg" > "$WORKDIR_FILE"
    fi
    prev="$arg"
done
exit 0
FAKE
    chmod +x "$TEST_TMP/fake-tts"
    printf 'CLI-AUDIO' > "$TEST_TMP/expected.bin"

    export VOICES_DIR="$VOICES"
    export TTS_BACKEND=llama-tts-cli
    export LLAMA_TTS_BIN="$TEST_TMP/fake-tts"
    export LLAMA_TTS_MODEL="$TEST_TMP/model.bin"
    export ARGS_FILE="$TEST_TMP/argv.txt"
    export WORKDIR_FILE="$TEST_TMP/workdir.txt"
    start_app
    speech '{"model":"tts-1","input":"narrate this","voice":"narrator1"}'
    [ "$output" = "200" ] || { cat "$TEST_TMP/app.log"; false; }
    grep -qi '^content-type: audio/wav' "$TEST_TMP/head.txt"
    cmp "$TEST_TMP/out.bin" "$TEST_TMP/expected.bin"
    # The speaker file must be the pack's reference under VOICES_DIR — that is
    # the whole product; anything else and the clone never happens.
    speaker="$(awk 'p { print; exit } $0 == "--tts-speaker-file" { p = 1 }' "$TEST_TMP/argv.txt")"
    [ "$speaker" = "$VOICES/narrator1/reference.wav" ]
    prompt="$(awk 'p { print; exit } $0 == "-p" { p = 1 }' "$TEST_TMP/argv.txt")"
    [ "$prompt" = "narrate this" ]
    grep -qx -- '--tts-lang' "$TEST_TMP/argv.txt"
    # A per-request temp dir that survives is how a TTS box fills its /tmp.
    workdir="$(cat "$TEST_TMP/workdir.txt")"
    [ -n "$workdir" ]
    [ ! -d "$workdir" ]
}

@test "response_format=mp3 ships mp3, converted from the engine's wav" {
    arm_omni
    # A fake ffmpeg on PATH: whatever wav arrives on stdin is tapped (so the
    # conversion provably ran on the engine's audio) and mp3 bytes come out.
    mkdir -p "$TEST_TMP/bin"
    cat > "$TEST_TMP/bin/ffmpeg" <<'FF'
#!/usr/bin/env bash
cat > "$STDIN_TAP"
printf 'MP3-BYTES'
FF
    chmod +x "$TEST_TMP/bin/ffmpeg"
    export STDIN_TAP="$TEST_TMP/ffmpeg-in.wav"
    # The fake must reach the APP's PATH: the shim runs `ffmpeg` by name.
    export PATH="$TEST_TMP/bin:$PATH"
    start_app
    speech '{"model":"tts-1","input":"hello","voice":"narrator1","response_format":"mp3"}'
    [ "$output" = "200" ] || { cat "$TEST_TMP/app.log"; false; }
    grep -qi '^content-type: audio/mpeg' "$TEST_TMP/head.txt"
    printf 'MP3-BYTES' > "$TEST_TMP/expected-mp3.bin"
    cmp "$TEST_TMP/out.bin" "$TEST_TMP/expected-mp3.bin"
    cmp "$TEST_TMP/ffmpeg-in.wav" "$TEST_TMP/upstream.wav"
    # The upstream is still asked for wav: mp3 is this shim's re-encode.
    python3 - "$TEST_TMP/capture.json" <<'PY'
import json, sys
assert json.load(open(sys.argv[1]))['response_format'] == 'wav'
PY
}

@test "the shipped env file boots the shim with its declared backend" {
    # Not a grep of the env file: source it the way systemd's EnvironmentFile
    # would, point the two machine-local values at the sandbox, and require the
    # endpoint to actually speak. A typo, a shell-ism or an expansion systemd
    # cannot parse shows up here as a dead process, not a green grep.
    set -a
    # shellcheck disable=SC1090
    source "$ENV_FILE"
    set +a
    arm_omni
    start_app
    speech '{"model":"tts-1","input":"hello","voice":"narrator1"}'
    [ "$output" = "200" ] || { cat "$TEST_TMP/app.log"; false; }
    cmp "$TEST_TMP/out.bin" "$TEST_TMP/upstream.wav"
}

@test "the quadlet unit keeps boot persistence and a read-only voice mount" {
    grep -qE '^Restart=always$' "$UNIT"
    grep -qE '^RestartSec=[0-9]+$' "$UNIT"
    grep -qE '^WantedBy=default.target$' "$UNIT"
    grep -qE '^PublishPort=([0-9.]+:)?7772:7772$' "$UNIT"
    grep -qE '^Volume=/home/dank/voices:/voices:ro$' "$UNIT"
}