buzz/tests/plugins/plugin_system_test.py

import os
import textwrap

import pytest

from buzz.plugins import loader
from buzz.plugins.loader import PluginLoadError, load_plugin_from_dir
from buzz.plugins.manager import PluginManager
from buzz.settings.settings import Settings
from buzz.transcriber.transcriber import Segment


VALID_PLUGIN = textwrap.dedent(
    """
    from buzz.plugins.base import (
        BuzzPlugin, PluginMetadata, ConfigField, ConfigFieldType,
    )

    class MyPlugin(BuzzPlugin):
        metadata = PluginMetadata(
            id="my_plugin",
            name="My Plugin",
            config_fields=[
                ConfigField(key="text", label="Text", default="hello"),
                ConfigField(key="flag", label="Flag",
                            type=ConfigFieldType.BOOL, default=True),
                ConfigField(key="secret", label="Secret",
                            type=ConfigFieldType.PASSWORD),
            ],
        )

        def before_transcription(self, task, context):
            return task.file_path + ".processed"

        def after_transcription(self, task, segments, context):
            return segments + [__import__("buzz.transcriber.transcriber",
                fromlist=["Segment"]).Segment(0, 1, "added")]

        def on_complete(self, transcription_id, task, segments, context):
            context.config["_ran"] = True
    """
)


def _write_plugin(plugin_dir, source=VALID_PLUGIN):
    os.makedirs(plugin_dir, exist_ok=True)
    with open(os.path.join(plugin_dir, "plugin.py"), "w") as f:
        f.write(source)
    return plugin_dir


@pytest.fixture()
def isolated_plugins(tmp_path, monkeypatch):
    """Point the loader at temp dirs and stub keyring + bundled-copy."""
    plugins_dir = tmp_path / "plugins"
    deps_dir = tmp_path / "deps"
    plugins_dir.mkdir()
    deps_dir.mkdir()

    monkeypatch.setattr(loader, "get_plugins_dir", lambda: str(plugins_dir))
    monkeypatch.setattr(loader, "get_plugins_deps_dir", lambda: str(deps_dir))
    monkeypatch.setattr(loader, "copy_bundled_plugins", lambda: None)

    secrets = {}
    import buzz.plugins.manager as manager_module

    monkeypatch.setattr(
        manager_module.keyring_store, "get_secret", lambda name: secrets.get(name, "")
    )
    monkeypatch.setattr(
        manager_module.keyring_store,
        "set_secret",
        lambda name, value: secrets.__setitem__(name, value),
    )
    monkeypatch.setattr(
        manager_module.keyring_store,
        "delete_secret",
        lambda name: secrets.pop(name, None),
    )
    return plugins_dir, deps_dir, secrets


@pytest.fixture()
def plugin_settings():
    """An isolated Settings instance so tests don't pollute real QSettings."""
    settings = Settings(application="plugins-test")
    # Don't fall back to the shared org-level settings file.
    settings.settings.setFallbacksEnabled(False)
    settings.settings.clear()
    settings.settings.sync()
    yield settings
    settings.settings.clear()
    settings.settings.sync()


def test_load_valid_plugin(tmp_path):
    plugin_dir = _write_plugin(str(tmp_path / "my_plugin"))
    plugin = load_plugin_from_dir(plugin_dir)
    assert plugin.metadata.id == "my_plugin"
    assert plugin.metadata.name == "My Plugin"


def test_load_invalid_plugin_no_subclass(tmp_path):
    plugin_dir = _write_plugin(str(tmp_path / "bad"), source="x = 1\n")
    with pytest.raises(PluginLoadError):
        load_plugin_from_dir(plugin_dir)


def test_manager_discovers_and_orders(qtbot, isolated_plugins, transcription_service, plugin_settings):
    plugins_dir, _deps, _secrets = isolated_plugins
    _write_plugin(str(plugins_dir / "my_plugin"))

    manager = PluginManager(transcription_service, plugin_settings)
    manager.initialize()

    assert "my_plugin" in manager.plugins
    assert "my_plugin" in manager.order


def test_config_persistence_and_password(qtbot, isolated_plugins, transcription_service, plugin_settings):
    plugins_dir, _deps, secrets = isolated_plugins
    _write_plugin(str(plugins_dir / "my_plugin"))

    manager = PluginManager(transcription_service, plugin_settings)
    manager.initialize()

    # Defaults come through.
    cfg = manager.get_config("my_plugin")
    assert cfg["text"] == "hello"
    assert cfg["flag"] is True

    manager.set_config(
        "my_plugin", {"text": "world", "flag": False, "secret": "s3cr3t"}
    )

    cfg = manager.get_config("my_plugin")
    assert cfg["text"] == "world"
    assert cfg["flag"] is False
    # Password stored via keyring, not QSettings.
    assert cfg["secret"] == "s3cr3t"
    assert secrets["plugin:my_plugin:secret"] == "s3cr3t"

    # Cleanup settings pollution.
    manager.remove("my_plugin")


def test_enable_and_order(qtbot, isolated_plugins, transcription_service, plugin_settings):
    plugins_dir, _deps, _secrets = isolated_plugins
    _write_plugin(str(plugins_dir / "a"))
    _write_plugin(
        str(plugins_dir / "b"),
        source=VALID_PLUGIN.replace('"my_plugin"', '"b_plugin"').replace(
            "MyPlugin", "BPlugin"
        ),
    )

    manager = PluginManager(transcription_service, plugin_settings)
    manager.initialize()

    assert manager.enabled_plugins_in_order() == []

    for pid in manager.order:
        manager.set_enabled(pid, True)
    assert len(manager.enabled_plugins_in_order()) == 2

    first = manager.order[0]
    manager.move(first, 1)
    assert manager.order[1] == first

    for pid in list(manager.plugins.keys()):
        manager.remove(pid)


def test_ai_summary_on_complete_writes_notes(monkeypatch):
    from buzz.plugins.ai_summary import plugin as ai_summary
    from buzz.plugins.base import PluginContext

    captured = {}

    class _Message:
        content = "A short summary."

    class _Choice:
        message = _Message()

    class _Completion:
        choices = [_Choice()]

    class _FakeClient:
        def __init__(self, *args, **kwargs):
            self.chat = self

        @property
        def completions(self):
            return self

        def create(self, *args, **kwargs):
            return _Completion()

    monkeypatch.setattr(ai_summary, "OpenAI", _FakeClient, raising=False)
    # Patch the import inside _summarize.
    import openai

    monkeypatch.setattr(openai, "OpenAI", _FakeClient)

    class _Service:
        def update_transcription_notes(self, tid, notes):
            captured["notes"] = (tid, notes)

    class _Task:
        file_path = "/tmp/a.wav"
        original_file_path = "/tmp/a.wav"

    context = PluginContext(
        config={
            "api_key": "key",
            "api_url": "https://example.com/v1",
            "model": "gpt-4o-mini",
            "prompt": "Summarise",
            "save_to_notes": True,
            "save_to_file": False,
        },
        transcription_service=_Service(),
        settings=None,
        logger=__import__("logging").getLogger("test"),
    )

    plugin = ai_summary.AISummaryPlugin()
    plugin.on_complete("tid-1", _Task(), [Segment(0, 1, "hello world")], context)

    assert captured["notes"] == ("tid-1", "A short summary.")


def test_transcript_resizer_loads():
    from buzz.plugins.loader import load_plugin_from_dir

    plugin = load_plugin_from_dir("buzz/plugins/transcript_resizer")
    assert plugin.metadata.id == "transcript_resizer"
    keys = [f.key for f in plugin.metadata.config_fields]
    assert "merge_by_gap" in keys and "max_length" in keys


def test_transcript_resizer_skips_without_word_timings():
    from buzz.plugins.transcript_resizer import plugin as resizer
    from buzz.plugins.base import PluginContext
    import logging

    replaced = {}

    class _Service:
        def replace_transcription_segments(self, tid, segs):
            replaced["called"] = True

    class _Options:
        word_level_timings = False

    class _Task:
        transcription_options = _Options()
        file_path = "/tmp/a.wav"
        original_file_path = "/tmp/a.wav"

    ctx = PluginContext(
        config={}, transcription_service=_Service(), settings=None,
        logger=logging.getLogger("test"),
    )
    resizer.TranscriptResizerPlugin().on_complete("tid", _Task(), [], ctx)
    assert "called" not in replaced


def test_transcript_resizer_regroups_with_word_timings(tmp_path, monkeypatch):
    from buzz.plugins.transcript_resizer import plugin as resizer
    from buzz.plugins.base import PluginContext
    from buzz.transcriber.transcriber import Segment
    import logging

    # A real (silent) audio file so load_audio succeeds.
    audio = tmp_path / "a.wav"
    audio.write_bytes(b"")  # load_audio is mocked, content irrelevant

    captured = {}

    class _DBSeg:
        def __init__(self, text, start, end):
            self.text = text
            self.start_time = start
            self.end_time = end

    class _Service:
        def get_transcription_segments(self, transcription_id):
            return [_DBSeg("Hello.", 0, 100), _DBSeg("World.", 100, 200)]

        def replace_transcription_segments(self, tid, segs):
            captured["segs"] = segs

    class _Options:
        word_level_timings = True
        language = "en"

    class _Task:
        transcription_options = _Options()
        file_path = str(audio)
        original_file_path = str(audio)

    # Mock the heavy audio + stable_whisper machinery.
    import buzz.whisper_audio as wa
    monkeypatch.setattr(wa, "load_audio", lambda path: b"audio")

    class _ResSeg:
        def __init__(self, start, end, text):
            self.start = start
            self.end = end
            self.text = text

    class _Result:
        segments = [_ResSeg(0.0, 2.0, "Hello. World.")]

    import stable_whisper
    monkeypatch.setattr(
        stable_whisper, "transcribe_any",
        lambda func, **kw: (func(kw.get("audio")), _Result())[1],
    )

    ctx = PluginContext(
        config={"merge_by_gap": True, "merge_gap_seconds": "0.2",
                "split_by_max_length": True, "max_length": "42",
                "split_by_punctuation": True, "punctuation": ".?!"},
        transcription_service=_Service(), settings=None,
        logger=logging.getLogger("test"),
    )
    resizer.TranscriptResizerPlugin().on_complete(
        "tid", _Task(), [Segment(0, 100, "Hello.")], ctx
    )

    assert "segs" in captured
    assert len(captured["segs"]) == 1
    assert captured["segs"][0].text == "Hello. World."
    assert captured["segs"][0].start == 0
    assert captured["segs"][0].end == 2000  # 2.0s * 100


def test_export_docx_loads():
    from buzz.plugins.loader import load_plugin_from_dir

    plugin = load_plugin_from_dir("buzz/plugins/export_docx")
    assert plugin.metadata.id == "export_docx"
    # The DOCX is built from the standard library; no third-party deps.
    assert plugin.metadata.pip_dependencies == []


def test_export_docx_writes_file(tmp_path):
    import logging
    from buzz.plugins.export_docx import plugin as ed
    from buzz.plugins.base import PluginContext

    class _DBSeg:
        def __init__(self, text, start, end):
            self.text = text
            self.start_time = start
            self.end_time = end

    class _Service:
        def get_transcription_segments(self, transcription_id):
            return [_DBSeg("Hello world.", 0, 1500), _DBSeg("Second part.", 5000, 7000)]

    class _Task:
        file_path = "/tmp/myaudio.wav"
        original_file_path = "/tmp/myaudio.wav"

    ctx = PluginContext(
        config={"output_folder": str(tmp_path), "include_timestamps": False},
        transcription_service=_Service(),
        settings=None,
        logger=logging.getLogger("test"),
    )
    ed.ExportDocxPlugin().on_complete("tid", _Task(), [], ctx)

    out = tmp_path / "myaudio.docx"
    assert out.exists() and out.stat().st_size > 0

    import zipfile

    with zipfile.ZipFile(out) as docx:
        assert "word/document.xml" in docx.namelist()
        document = docx.read("word/document.xml").decode("utf-8")

    assert "myaudio" in document  # heading uses the file stem
    assert "Hello world." in document


def test_plugins_dialog_builds_and_wraps(qtbot):
    from PyQt6.QtCore import Qt
    from buzz.widgets.plugins_dialog.plugins_dialog import PluginsDialog
    from buzz.plugins.base import PluginMetadata

    class _FakePlugin:
        metadata = PluginMetadata(
            id="demo",
            name="Demo Plugin",
            description="A long description " * 20,
        )

    class _FakeManager:
        order = ["demo"]

        def all_plugins_in_order(self):
            return [_FakePlugin()]

        def is_enabled(self, _pid):
            return True

    dialog = PluginsDialog(_FakeManager())
    qtbot.addWidget(dialog)

    assert dialog.list_widget.wordWrap() is True
    assert (
        dialog.list_widget.horizontalScrollBarPolicy()
        == Qt.ScrollBarPolicy.ScrollBarAlwaysOff
    )
    assert dialog.list_widget.count() == 1
    # Default size is ~50% wider than the original 560px.
    assert dialog.width() >= 800


def test_plugin_gettext_translates(tmp_path, monkeypatch):
    import json

    import buzz.plugins.base as base

    plugin_dir = tmp_path / "p"
    locale_dir = plugin_dir / "locale"
    locale_dir.mkdir(parents=True)
    (locale_dir / "lv_LV.json").write_text(
        json.dumps({"Hello": "Sveiki"}), encoding="utf-8"
    )
    plugin_file = str(plugin_dir / "plugin.py")

    monkeypatch.setattr(base, "_current_locale", lambda: "lv_LV")
    translate = base.plugin_gettext(plugin_file)
    assert translate("Hello") == "Sveiki"
    # Untranslated strings fall through unchanged.
    assert translate("Goodbye") == "Goodbye"


def test_plugin_gettext_falls_back_to_language(tmp_path, monkeypatch):
    import json

    import buzz.plugins.base as base

    plugin_dir = tmp_path / "p"
    locale_dir = plugin_dir / "locale"
    locale_dir.mkdir(parents=True)
    (locale_dir / "lv.json").write_text(
        json.dumps({"Hello": "Sveiki"}), encoding="utf-8"
    )

    monkeypatch.setattr(base, "_current_locale", lambda: "lv_LV")
    translate = base.plugin_gettext(str(plugin_dir / "plugin.py"))
    assert translate("Hello") == "Sveiki"


def test_plugin_gettext_no_locale_dir(tmp_path, monkeypatch):
    import buzz.plugins.base as base

    monkeypatch.setattr(base, "_current_locale", lambda: "lv_LV")
    translate = base.plugin_gettext(str(tmp_path / "plugin.py"))
    assert translate("Anything") == "Anything"


def _eld_plugin():
    from buzz.plugins.enhanced_language_detection import plugin as eld

    return eld.EnhancedLanguageDetectionPlugin()


def _eld_context(config=None):
    import logging
    from buzz.plugins.base import PluginContext

    captured = {}

    class _Service:
        def update_transcription_language(self, tid, language):
            captured["lang"] = (tid, language)

    ctx = PluginContext(
        config=config or {},
        transcription_service=_Service(),
        settings=None,
        logger=logging.getLogger("test"),
    )
    return ctx, captured


class _EldOptions:
    def __init__(self, language):
        self.language = language


class _EldTask:
    def __init__(self, language, file_path="/tmp/a.wav", uid="tid-1"):
        self.transcription_options = _EldOptions(language)
        self.file_path = file_path
        self.uid = uid


def test_enhanced_language_detection_loads():
    plugin = load_plugin_from_dir("buzz/plugins/enhanced_language_detection")
    assert plugin.metadata.id == "enhanced_language_detection"
    keys = [f.key for f in plugin.metadata.config_fields]
    assert "download_tiny_if_missing" in keys


def test_eld_skips_explicit_language(monkeypatch):
    plugin = _eld_plugin()
    ctx, captured = _eld_context()

    # Detection must never be invoked for an explicit, non-default language.
    monkeypatch.setattr(
        plugin, "_resolve_model_path", lambda c: (_ for _ in ()).throw(AssertionError())
    )

    task = _EldTask(language="de")
    plugin.before_transcription(task, ctx)

    assert task.transcription_options.language == "de"
    assert "lang" not in captured


@pytest.mark.parametrize("language", [None, "", "en", "EN"])
def test_eld_detects_when_auto_or_en(monkeypatch, language):
    plugin = _eld_plugin()
    ctx, captured = _eld_context()

    monkeypatch.setattr(plugin, "_resolve_model_path", lambda c: "/fake/model.bin")
    from buzz.transcriber import whisper_cpp

    monkeypatch.setattr(
        whisper_cpp.WhisperCpp, "detect_language", staticmethod(lambda f, m: "fr")
    )

    task = _EldTask(language=language)
    plugin.before_transcription(task, ctx)

    assert task.transcription_options.language == "fr"
    assert captured["lang"] == ("tid-1", "fr")


def test_eld_skips_without_file_path(monkeypatch):
    plugin = _eld_plugin()
    ctx, captured = _eld_context()
    monkeypatch.setattr(
        plugin, "_resolve_model_path", lambda c: (_ for _ in ()).throw(AssertionError())
    )

    task = _EldTask(language=None, file_path=None)
    plugin.before_transcription(task, ctx)

    assert task.transcription_options.language is None
    assert "lang" not in captured


def test_eld_keeps_language_when_no_detection(monkeypatch):
    plugin = _eld_plugin()
    ctx, captured = _eld_context()

    monkeypatch.setattr(plugin, "_resolve_model_path", lambda c: "/fake/model.bin")
    from buzz.transcriber import whisper_cpp

    monkeypatch.setattr(
        whisper_cpp.WhisperCpp, "detect_language", staticmethod(lambda f, m: None)
    )

    task = _EldTask(language=None)
    plugin.before_transcription(task, ctx)

    assert task.transcription_options.language is None
    assert "lang" not in captured


def test_eld_resolve_model_path_picks_largest_skips_custom_lumii(monkeypatch):
    from buzz import model_loader
    from buzz.model_loader import WhisperModelSize

    available = {
        WhisperModelSize.TINY: "/models/tiny.bin",
        WhisperModelSize.SMALL: "/models/small.bin",
        WhisperModelSize.LUMII: "/models/lumii.bin",
        WhisperModelSize.CUSTOM: "/models/custom.bin",
    }

    def fake_path(self):
        return available.get(self.whisper_model_size)

    monkeypatch.setattr(
        model_loader.TranscriptionModel, "get_local_model_path", fake_path
    )

    plugin = _eld_plugin()
    ctx, _ = _eld_context()
    # SMALL outranks TINY; LUMII/CUSTOM are excluded even though available.
    assert plugin._resolve_model_path(ctx) == "/models/small.bin"


def test_eld_resolve_model_path_returns_none_when_download_disabled(monkeypatch):
    from buzz import model_loader

    monkeypatch.setattr(
        model_loader.TranscriptionModel, "get_local_model_path", lambda self: None
    )

    plugin = _eld_plugin()
    ctx, _ = _eld_context(config={"download_tiny_if_missing": False})
    assert plugin._resolve_model_path(ctx) is None


def test_eld_resolve_model_path_downloads_tiny_when_missing(monkeypatch):
    from buzz import model_loader
    from buzz.model_loader import WhisperModelSize

    state = {"downloaded": False}

    def fake_path(self):
        if state["downloaded"] and self.whisper_model_size == WhisperModelSize.TINY:
            return "/models/tiny.bin"
        return None

    class _FakeDownloader:
        def __init__(self, model=None):
            self.model = model

        def run(self):
            state["downloaded"] = True

    monkeypatch.setattr(
        model_loader.TranscriptionModel, "get_local_model_path", fake_path
    )
    monkeypatch.setattr(model_loader, "ModelDownloader", _FakeDownloader)

    plugin = _eld_plugin()
    ctx, _ = _eld_context(config={"download_tiny_if_missing": True})
    assert plugin._resolve_model_path(ctx) == "/models/tiny.bin"
    assert state["downloaded"] is True


def test_before_and_after_hooks(qtbot, isolated_plugins, transcription_service, plugin_settings):
    plugins_dir, _deps, _secrets = isolated_plugins
    _write_plugin(str(plugins_dir / "my_plugin"))

    manager = PluginManager(transcription_service, plugin_settings)
    manager.initialize()
    manager.set_enabled("my_plugin", True)

    class FakeTask:
        file_path = "/tmp/audio.wav"

    task = FakeTask()
    manager.run_before_transcription(task)
    assert task.file_path == "/tmp/audio.wav.processed"

    segments = [Segment(0, 1, "original")]
    result = manager.run_after_transcription(task, segments)
    assert len(result) == 2
    assert result[-1].text == "added"

    manager.remove("my_plugin")


# ---------------------------------------------------------------------------
# skip_already_transcribed plugin tests
# ---------------------------------------------------------------------------

def _sat_plugin():
    from buzz.plugins.skip_already_transcribed import plugin as sat
    return sat.SkipAlreadyTranscribedPlugin()


def _sat_context(config=None, db_records=None):
    import logging
    from buzz.plugins.base import PluginContext

    # db_records: {filename: [db_segment, ...]} where db_segment has .start_time/.end_time/.text
    records = db_records or {}

    class _DBSeg:
        def __init__(self, text, start, end):
            self.text = text
            self.start_time = start
            self.end_time = end

    class _Service:
        def find_completed_transcription_by_filename(self, filename):
            return filename if filename in records else None

        def get_transcription_segments(self, tid):
            return records.get(tid, [])

    ctx = PluginContext(
        config=config or {"check_result_files": True, "check_database": False},
        transcription_service=_Service(),
        settings=None,
        logger=logging.getLogger("test"),
    )
    return ctx


class _SatTask:
    def __init__(self, file_path="/tmp/audio.mp3", output_directory=None):
        self.file_path = file_path
        self.original_file_path = file_path
        self.output_directory = output_directory


def test_skip_already_transcribed_loads():
    from buzz.plugins.loader import load_plugin_from_dir
    plugin = load_plugin_from_dir("buzz/plugins/skip_already_transcribed")
    assert plugin.metadata.id == "skip_already_transcribed"
    keys = [f.key for f in plugin.metadata.config_fields]
    assert "check_result_files" in keys
    assert "check_database" in keys


def test_sat_returns_none_when_no_files_and_db_disabled(tmp_path):
    plugin = _sat_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")
    ctx = _sat_context(config={"check_result_files": True, "check_database": False})
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is None


def test_sat_skips_on_txt_file(tmp_path):
    plugin = _sat_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")
    txt = tmp_path / "audio (transcribed on 01-Jan-2024 12-00-00).txt"
    txt.write_text("Hello world.", encoding="utf-8")

    ctx = _sat_context(config={"check_result_files": True, "check_database": False})
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is not None
    assert len(result) == 1
    assert result[0].text == "Hello world."
    assert result[0].start == 0
    assert result[0].end == 0


def test_sat_skips_on_srt_file(tmp_path):
    plugin = _sat_plugin()
    audio = tmp_path / "lecture.mp3"
    audio.write_bytes(b"")
    srt = tmp_path / "lecture (transcribed on 01-Jan-2024 12-00-00).srt"
    srt.write_text(
        "1\n00:00:01,000 --> 00:00:03,500\nHello there.\n\n"
        "2\n00:00:04,000 --> 00:00:06,000\nGeneral Kenobi.\n\n",
        encoding="utf-8",
    )

    ctx = _sat_context(config={"check_result_files": True, "check_database": False})
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is not None
    assert len(result) == 2
    assert result[0].text == "Hello there."
    assert result[0].start == 1000
    assert result[0].end == 3500
    assert result[1].text == "General Kenobi."
    assert result[1].start == 4000


def test_sat_skips_on_vtt_file(tmp_path):
    plugin = _sat_plugin()
    audio = tmp_path / "talk.mp3"
    audio.write_bytes(b"")
    vtt = tmp_path / "talk (transcribed on 01-Jan-2024 12-00-00).vtt"
    vtt.write_text(
        "WEBVTT\n\n"
        "00:00:00.500 --> 00:00:02.000\nFirst line.\n\n"
        "00:00:03.000 --> 00:00:05.000\nSecond line.\n\n",
        encoding="utf-8",
    )

    ctx = _sat_context(config={"check_result_files": True, "check_database": False})
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is not None
    assert len(result) == 2
    assert result[0].text == "First line."
    assert result[0].start == 500
    assert result[1].text == "Second line."
    assert result[1].start == 3000


def test_sat_returns_none_when_result_files_disabled(tmp_path):
    plugin = _sat_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")
    (tmp_path / "audio.txt").write_text("Hello world.", encoding="utf-8")

    ctx = _sat_context(config={"check_result_files": False, "check_database": False})
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is None


def test_sat_skips_on_db_record(tmp_path):
    plugin = _sat_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")

    class _DBSeg:
        def __init__(self, text, start, end):
            self.text = text
            self.start_time = start
            self.end_time = end

    ctx = _sat_context(
        config={"check_result_files": False, "check_database": True},
        db_records={"audio.mp3": [_DBSeg("From DB.", 0, 2000)]},
    )
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is not None
    assert len(result) == 1
    assert result[0].text == "From DB."
    assert result[0].start == 0
    assert result[0].end == 2000


def test_sat_no_skip_when_db_returns_none(tmp_path):
    plugin = _sat_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")

    ctx = _sat_context(
        config={"check_result_files": False, "check_database": True},
        db_records={},
    )
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is None


def test_sat_result_files_checked_before_db(tmp_path):
    """Result file check wins if both options are enabled."""
    plugin = _sat_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")
    (tmp_path / "audio (transcribed on 01-Jan-2024 12-00-00).txt").write_text(
        "From file.", encoding="utf-8"
    )

    class _DBSeg:
        text = "From DB."
        start_time = 0
        end_time = 1000

    ctx = _sat_context(
        config={"check_result_files": True, "check_database": True},
        db_records={"audio.mp3": [_DBSeg()]},
    )
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is not None
    assert result[0].text == "From file."


def test_sat_strips_speech_suffix(tmp_path):
    """Files with _speech suffix (post speech-extraction) match by original stem."""
    plugin = _sat_plugin()
    audio = tmp_path / "audio_speech.mp3"
    audio.write_bytes(b"")
    (tmp_path / "audio (transcribed on 01-Jan-2024 12-00-00).txt").write_text(
        "From file.", encoding="utf-8"
    )

    ctx = _sat_context(config={"check_result_files": True, "check_database": False})
    result = plugin.check_skip(_SatTask(file_path=str(audio)), ctx)
    assert result is not None
    assert result[0].text == "From file."


def test_sat_run_check_skip_integration(qtbot, isolated_plugins, transcription_service, plugin_settings, tmp_path):
    """Manager.run_check_skip calls through to the plugin."""
    from buzz.plugins.manager import PluginManager
    from buzz.plugins import loader

    plugins_dir, _deps, _secrets = isolated_plugins

    import shutil
    sat_src = "buzz/plugins/skip_already_transcribed"
    shutil.copytree(sat_src, str(plugins_dir / "skip_already_transcribed"))

    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")
    (tmp_path / "audio (transcribed on 01-Jan-2024 12-00-00).txt").write_text(
        "Existing transcript.", encoding="utf-8"
    )

    manager = PluginManager(transcription_service, plugin_settings)
    manager.initialize()
    manager.set_enabled("skip_already_transcribed", True)
    manager.set_config("skip_already_transcribed", {"check_result_files": True, "check_database": False})

    task = _SatTask(file_path=str(audio))
    should_skip, segments = manager.run_check_skip(task)

    assert should_skip is True
    assert len(segments) == 1
    assert segments[0].text == "Existing transcript."

    manager.remove("skip_already_transcribed")


# ---------------------------------------------------------------------------
# DeepFilterNet plugin tests
# ---------------------------------------------------------------------------

def _dfn_plugin():
    from buzz.plugins.loader import load_plugin_from_dir
    return load_plugin_from_dir("buzz/plugins/deep_filter_net")


def _dfn_context(config=None):
    import logging
    from buzz.plugins.base import PluginContext
    return PluginContext(
        config=config or {},
        transcription_service=None,
        settings=None,
        logger=logging.getLogger("test"),
    )


class _DfnTask:
    def __init__(self, file_path="/tmp/audio.mp3"):
        self.file_path = file_path
        self.original_file_path = file_path


def test_deep_filter_net_loads():
    plugin = _dfn_plugin()
    assert plugin.metadata.id == "deep_filter_net"
    keys = [f.key for f in plugin.metadata.config_fields]
    assert "keep_denoised_file" in keys


def test_deep_filter_net_before_transcription_creates_denoised_file(tmp_path):
    plugin = _dfn_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")

    import sys
    from unittest.mock import MagicMock, patch

    mock_audio_tensor = MagicMock()
    mock_sr = MagicMock()
    mock_model = MagicMock()
    mock_df_state = MagicMock()
    mock_df_state.sr.return_value = mock_sr
    mock_enhanced = MagicMock()

    df_module = MagicMock()
    df_module.init_df.return_value = (mock_model, mock_df_state, None)
    df_module.load_audio.return_value = (mock_audio_tensor, mock_sr)
    df_module.enhance.return_value = mock_enhanced

    with patch.dict(sys.modules, {"df": MagicMock(), "df.enhance": df_module}):
        task = _DfnTask(file_path=str(audio))
        result = plugin.before_transcription(task, _dfn_context())

    expected = str(tmp_path / "audio_DeepFilterNet3.wav")
    assert result == expected
    df_module.init_df.assert_called_once()
    df_module.load_audio.assert_called_once_with(str(audio), sr=mock_sr)
    df_module.enhance.assert_called_once_with(mock_model, mock_df_state, mock_audio_tensor)
    df_module.save_audio.assert_called_once_with(expected, mock_enhanced, mock_sr)


def test_deep_filter_net_returns_none_on_error():
    plugin = _dfn_plugin()

    import sys
    from unittest.mock import MagicMock, patch

    df_module = MagicMock()
    df_module.init_df.side_effect = RuntimeError("model not found")

    with patch.dict(sys.modules, {"df": MagicMock(), "df.enhance": df_module}):
        task = _DfnTask(file_path="/tmp/audio.mp3")
        result = plugin.before_transcription(task, _dfn_context())

    assert result is None


def test_deep_filter_net_deletes_file_when_keep_false(tmp_path):
    plugin = _dfn_plugin()
    denoised = tmp_path / "audio_DeepFilterNet3.wav"
    denoised.write_bytes(b"")

    task = _DfnTask(file_path=str(denoised))
    plugin.on_complete(None, task, [], _dfn_context(config={"keep_denoised_file": False}))

    assert not denoised.exists()


def test_deep_filter_net_keeps_file_when_keep_true(tmp_path):
    plugin = _dfn_plugin()
    denoised = tmp_path / "audio_DeepFilterNet3.wav"
    denoised.write_bytes(b"")

    task = _DfnTask(file_path=str(denoised))
    plugin.on_complete(None, task, [], _dfn_context(config={"keep_denoised_file": True}))

    assert denoised.exists()


def test_deep_filter_net_does_not_delete_non_dfn_file(tmp_path):
    plugin = _dfn_plugin()
    audio = tmp_path / "audio.mp3"
    audio.write_bytes(b"")

    task = _DfnTask(file_path=str(audio))
    plugin.on_complete(None, task, [], _dfn_context(config={"keep_denoised_file": False}))

    assert audio.exists()