UmbraAudio/tests/test_delete_speaker.py

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Tests for speaker deletion.

The speakers page has offered a Delete button for a long time. The endpoint
it called, ``DELETE /api/speakers/{index}``, did not exist: htmx sent the
request and the server answered ``405 Method Not Allowed``, so the button
silently did nothing.

Deleting a speaker who still has dialogue is also destructive in a way that
is easy to miss - the utterances keep a ``speaker_name`` that no longer
resolves - so the endpoint refuses unless the caller says where that dialogue
should go.
"""
from __future__ import annotations

import json
import sqlite3
from collections.abc import Iterator
from pathlib import Path
from typing import Any

import pytest
from fastapi.testclient import TestClient

SPEAKERS = [
    {"primary_name": "Alpha", "aliases": ["A-one"], "vocal_traits": ["calm"]},
    {"primary_name": "Beta", "aliases": [], "vocal_traits": ["deep"]},
    {"primary_name": "Silent", "aliases": [], "vocal_traits": []},
]


@pytest.fixture()
def data_dir(tmp_path: Path) -> Iterator[Path]:
    """A throwaway project, wired in as the active one."""
    from umbraaudio.config import AppConfig

    d = tmp_path / "proj" / ".umbra"
    d.mkdir(parents=True)
    (d / "speakers.json").write_text(json.dumps({"speakers": SPEAKERS}))

    # Use the app's own schema: _get_db runs SCHEMA_SQL on connect, and a
    # hand-rolled minimal table fails against it.
    from umbraaudio.web.routes import SCHEMA_SQL

    conn = sqlite3.connect(str(d / "project.db"))
    conn.executescript(SCHEMA_SQL)
    conn.execute(
        "INSERT INTO utterances (chapter_id, utterance_index, speaker_name, text, type) "
        "VALUES (1, 0, 'Beta', 'hello', 'dialogue')"
    )
    conn.execute(
        "INSERT INTO utterances (chapter_id, utterance_index, speaker_name, text, type) "
        "VALUES (1, 1, 'A-one', 'via alias', 'dialogue')"
    )
    conn.commit()
    conn.close()

    cfg = AppConfig()
    cfg.project_dir = tmp_path / "proj"
    cfg.data_dir = d

    import umbraaudio.config as config_mod

    original = config_mod.load_config
    config_mod.load_config = lambda _slug=None: cfg  # type: ignore[assignment]
    try:
        yield d
    finally:
        config_mod.load_config = original  # type: ignore[assignment]


@pytest.fixture()
def client(data_dir: Path, monkeypatch: pytest.MonkeyPatch) -> Iterator[TestClient]:
    from umbraaudio.config import AppConfig
    from umbraaudio.web import create_app

    cfg = AppConfig()
    cfg.project_dir = data_dir.parent
    cfg.data_dir = data_dir

    # The app is gated by fail-closed auth middleware, so an unauthenticated
    # request is redirected to /login and TestClient follows it - which would
    # turn every assertion into a check against the login page. Mint a real
    # session with a pinned secret so the tests neither touch the production
    # .umbra/session_secret nor need the real password.
    import umbraaudio.auth as auth_mod

    monkeypatch.setattr(auth_mod, "_session_secret", lambda: b"k" * 32)
    token = auth_mod.issue_session(auth_mod.USERNAME)

    # create_app caches the app in a module-level singleton, so it must be
    # cleared between tests or the second test would reuse the first test's
    # config - and therefore the first test's temporary directory.
    import umbraaudio.web as web_mod

    web_mod._app = None
    try:
        c = TestClient(create_app(cfg), follow_redirects=False)
        c.cookies.set("active_project", "testproj")
        c.cookies.set(auth_mod.COOKIE_NAME, token)
        yield c
    finally:
        web_mod._app = None


def _spk(data_dir: Path) -> list[dict[str, Any]]:
    return json.loads((data_dir / "speakers.json").read_text())["speakers"]


def _names(data_dir: Path) -> list[str]:
    return [s["primary_name"] for s in _spk(data_dir)]


def _rows(data_dir: Path) -> dict[str, str]:
    conn = sqlite3.connect(str(data_dir / "project.db"))
    rows = dict(conn.execute("SELECT text, speaker_name FROM utterances"))
    conn.close()
    return rows


def test_delete_speaker_without_dialogue_succeeds(
    client: TestClient, data_dir: Path
) -> None:
    idx = _names(data_dir).index("Silent")
    assert client.delete(f"/api/speakers/{idx}").status_code == 200
    assert "Silent" not in _names(data_dir)


def test_delete_speaker_with_dialogue_is_refused(
    client: TestClient, data_dir: Path
) -> None:
    """Otherwise the utterances keep a name that no longer resolves."""
    idx = _names(data_dir).index("Beta")
    r = client.delete(f"/api/speakers/{idx}")
    assert r.status_code == 409
    body = r.json()
    assert body["utterance_count"] == 1
    assert "Beta" in body["error"]
    assert "Beta" in _names(data_dir), "must not be removed"


def test_refusal_offers_the_other_speakers_as_targets(
    client: TestClient, data_dir: Path
) -> None:
    idx = _names(data_dir).index("Beta")
    options = client.delete(f"/api/speakers/{idx}").json()["options"]
    assert "Alpha" in options
    assert "Beta" not in options, "a speaker cannot be reassigned to itself"


def test_reassigning_moves_the_dialogue(client: TestClient, data_dir: Path) -> None:
    idx = _names(data_dir).index("Beta")
    assert client.delete(f"/api/speakers/{idx}?reassign_to=Alpha").status_code == 200
    assert _rows(data_dir)["hello"] == "Alpha"
    assert "Beta" not in _names(data_dir)


def test_reassignment_keeps_the_old_name_as_an_alias(
    client: TestClient, data_dir: Path
) -> None:
    """So anything still referring to the deleted name keeps resolving."""
    idx = _names(data_dir).index("Beta")
    client.delete(f"/api/speakers/{idx}?reassign_to=Alpha")
    alpha = next(s for s in _spk(data_dir) if s["primary_name"] == "Alpha")
    assert "Beta" in alpha["aliases"]


def test_dialogue_stored_under_an_alias_also_counts(
    client: TestClient, data_dir: Path
) -> None:
    """Utterances are frequently stored under an alias, not the primary name."""
    idx = _names(data_dir).index("Alpha")
    r = client.delete(f"/api/speakers/{idx}")
    assert r.status_code == 409
    assert r.json()["utterance_count"] == 1


def test_reassignment_target_matches_case_insensitively(
    client: TestClient, data_dir: Path
) -> None:
    idx = _names(data_dir).index("Alpha")
    assert client.delete(f"/api/speakers/{idx}?reassign_to=beta").status_code == 200
    assert _rows(data_dir)["via alias"] == "Beta"
    assert "Alpha" not in _names(data_dir)


def test_unknown_reassignment_target_is_rejected(
    client: TestClient, data_dir: Path
) -> None:
    idx = _names(data_dir).index("Beta")
    assert client.delete(f"/api/speakers/{idx}?reassign_to=Nobody").status_code == 400
    assert "Beta" in _names(data_dir), "must not delete"


def test_out_of_range_index_is_rejected(client: TestClient) -> None:
    assert client.delete("/api/speakers/999").status_code == 400


def test_deleting_leaves_other_speakers_untouched(
    client: TestClient, data_dir: Path
) -> None:
    idx = _names(data_dir).index("Silent")
    client.delete(f"/api/speakers/{idx}")
    alpha = next(s for s in _spk(data_dir) if s["primary_name"] == "Alpha")
    assert alpha["aliases"] == ["A-one"], "unrelated aliases must be untouched"