OBLITERATUS/tests/test_ui_watchtower_contracts.py

"""Pure handler contracts for the Watchtower and one-click Gradio tabs."""

from __future__ import annotations

import importlib
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock

import pytest

auto_obliterate = importlib.import_module("obliteratus.auto_obliterate")
ui_watchtower = importlib.import_module("obliteratus.ui_watchtower")


def test_component_update_delegates_to_gradio_when_available(monkeypatch):
    update = Mock(return_value={"delegated": True})
    monkeypatch.setattr(ui_watchtower, "gr", SimpleNamespace(update=update))

    assert ui_watchtower._component_update(interactive=True) == {"delegated": True}
    update.assert_called_once_with(interactive=True)


def test_trending_choices_use_watchtower_and_fall_back_on_empty_or_error(monkeypatch):
    watchtower = SimpleNamespace(get_model_choices=lambda: ["org/one", "org/two"])
    monkeypatch.setattr(ui_watchtower, "_get_watchtower", lambda: watchtower)
    assert ui_watchtower._get_trending_choices() == ["org/one", "org/two"]

    watchtower.get_model_choices = lambda: []
    assert ui_watchtower._get_trending_choices()[0] == "meta-llama/Llama-3.1-8B-Instruct"

    monkeypatch.setattr(
        ui_watchtower,
        "_get_watchtower",
        Mock(side_effect=RuntimeError("offline")),
    )
    assert "Qwen/Qwen3-4B" in ui_watchtower._get_trending_choices()


@pytest.mark.parametrize("model_id", ["", "   ", None])
def test_one_click_rejects_missing_model_without_starting_work(model_id):
    outputs = list(ui_watchtower._run_one_click(model_id, 3, 5))

    assert len(outputs) == 1
    assert "Please enter or select" in outputs[0][0]
    assert outputs[0][3]["interactive"] is False


def test_one_click_streams_progress_and_preserves_zero_refusal_rate(monkeypatch, tmp_path):
    output = tmp_path / "model"
    output.mkdir()
    created: list[object] = []
    final = SimpleNamespace(
        success=True,
        final_output_dir=str(output),
        final_refusal_rate=0.0,
        total_time_seconds=12.5,
    )

    class FakeObliterator:
        def __init__(self, **kwargs):
            created.append(kwargs)
            self._result = SimpleNamespace(final_output_dir=str(output))

        def run(self):
            yield "running", "step", "metrics"
            return final

        def _format_metrics(self):
            return "final metrics"

    monkeypatch.setattr(auto_obliterate, "AutoObliterator", FakeObliterator)

    outputs = list(ui_watchtower._run_one_click("  org/model  ", 3.9, 5))

    assert created == [
        {
            "model_id": "org/model",
            "max_iterations": 3,
            "target_refusal_rate": 0.05,
        }
    ]
    assert outputs[0][0] == "running"
    assert outputs[0][3]["interactive"] is True
    assert outputs[-1][0] == "✅ Complete!"
    assert "Final refusal rate: 0.0" in outputs[-1][1]
    assert "unknown" not in outputs[-1][1]


def test_one_click_converts_generator_failure_to_noninteractive_error(monkeypatch):
    class FailingObliterator:
        def __init__(self, **_kwargs):
            self._result = SimpleNamespace(final_output_dir=None)

        def run(self):
            yield "starting", "", ""
            raise RuntimeError("device failed")

    monkeypatch.setattr(auto_obliterate, "AutoObliterator", FailingObliterator)

    outputs = list(ui_watchtower._run_one_click("org/model", 1, 5))

    assert outputs[0][0] == "starting"
    assert "device failed" in outputs[-1][0]
    assert outputs[-1][3]["interactive"] is False


def test_download_result_selects_latest_numeric_iteration(monkeypatch, tmp_path):
    monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
    base = tmp_path / ".obliteratus" / "auto_obliterate" / "org_model"
    for name in ("iter_invalid", "iter_2", "iter_10"):
        directory = base / name
        directory.mkdir(parents=True)
        (directory / "config.json").write_text("{}", encoding="utf-8")

    assert ui_watchtower._download_result("org/model") == str(base / "iter_10")


def test_download_result_handles_empty_id_and_unreadable_iteration(monkeypatch, tmp_path):
    monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
    base = tmp_path / ".obliteratus" / "auto_obliterate" / "org_model"
    broken = base / "iter_20"
    broken.mkdir(parents=True)
    (broken / "config.json").write_text("{}", encoding="utf-8")
    valid = base / "iter_10"
    valid.mkdir()
    (valid / "config.json").write_text("{}", encoding="utf-8")
    resolve = Path.resolve

    def controlled_resolve(path, *args, **kwargs):
        if path == broken:
            raise OSError("unreadable")
        return resolve(path, *args, **kwargs)

    monkeypatch.setattr(Path, "resolve", controlled_resolve)

    assert ui_watchtower._download_result("") is None
    assert ui_watchtower._download_result("org/model") == str(valid)


def test_download_result_handles_unreadable_managed_model_directory(monkeypatch, tmp_path):
    monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
    base = tmp_path / ".obliteratus" / "auto_obliterate" / "org_model"
    base.mkdir(parents=True)
    resolve = Path.resolve

    def controlled_resolve(path, *args, **kwargs):
        if path == base:
            raise OSError("unreadable model directory")
        return resolve(path, *args, **kwargs)

    monkeypatch.setattr(Path, "resolve", controlled_resolve)

    assert ui_watchtower._download_result("org/model") is None


def test_download_result_rejects_symlink_that_escapes_managed_root(monkeypatch, tmp_path):
    monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
    base = tmp_path / ".obliteratus" / "auto_obliterate" / "org_model"
    valid = base / "iter_10"
    valid.mkdir(parents=True)
    (valid / "config.json").write_text("{}", encoding="utf-8")
    outside = tmp_path / "outside"
    outside.mkdir()
    (outside / "config.json").write_text("{}", encoding="utf-8")
    (base / "iter_99").symlink_to(outside, target_is_directory=True)

    assert ui_watchtower._download_result("../../org/model") is None
    assert ui_watchtower._download_result("org/model") == str(valid)


@pytest.mark.parametrize("model_id", [".", ".."])
def test_download_result_rejects_managed_root_aliases(monkeypatch, tmp_path, model_id):
    monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
    root = tmp_path / ".obliteratus" / "auto_obliterate"
    iteration = root / "iter_10"
    iteration.mkdir(parents=True)
    (iteration / "config.json").write_text("{}", encoding="utf-8")

    assert ui_watchtower._download_result(model_id) is None


def test_download_result_rejects_model_directory_symlink_escape(monkeypatch, tmp_path):
    monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path))
    root = tmp_path / ".obliteratus" / "auto_obliterate"
    root.mkdir(parents=True)
    outside = tmp_path / "outside"
    iteration = outside / "iter_99"
    iteration.mkdir(parents=True)
    (iteration / "config.json").write_text("{}", encoding="utf-8")
    (root / "org_model").symlink_to(outside, target_is_directory=True)

    assert ui_watchtower._download_result("org/model") is None


def test_scan_handler_returns_data_and_escaped_failure(monkeypatch):
    watchtower = SimpleNamespace(
        scan=lambda on_log: (on_log("scanned") or [SimpleNamespace()]),
        get_stats=lambda: {"total_tracked": 1},
        format_table=lambda: [["org/model"]],
    )
    monkeypatch.setattr(ui_watchtower, "_get_watchtower", lambda: watchtower)

    status, table, log = ui_watchtower._scan_now()

    assert "1" in status
    assert table == [["org/model"]]
    assert log == "scanned"

    monkeypatch.setattr(
        ui_watchtower,
        "_get_watchtower",
        Mock(side_effect=RuntimeError("<script>alert(1)</script>")),
    )
    status, table, log = ui_watchtower._scan_now()
    assert "<script>" not in status
    assert "&lt;script&gt;" in status
    assert table == []
    assert "<script>" not in log


def test_status_html_formats_zero_counts_and_escapes_invalid_timestamp():
    html = ui_watchtower._format_status_html(
        {
            "total_tracked": 0,
            "by_status": {"new": 0, "queued": 0, "obliterated": 0},
            "last_scan": "<img src=x onerror=alert(1)>",
            "scan_count": 0,
        },
        new_count=2,
    )

    assert "+2 NEW" in html
    assert "Scan #0" in html
    assert "<img" not in html
    assert "&lt;img" in html


def test_history_preserves_zero_metrics_and_handles_empty_or_failure(monkeypatch):
    model = SimpleNamespace(
        model_id="org/model",
        obliteration_metrics={"method": "advanced", "refusal_rate": 0.0, "perplexity": 0.0},
        last_updated="2026-08-16T10:30:00+00:00",
    )
    watchtower = SimpleNamespace(get_obliterated=lambda: [model])
    monkeypatch.setattr(ui_watchtower, "_get_watchtower", lambda: watchtower)

    history = ui_watchtower._get_obliteration_history()

    assert "| org/model | advanced | 0.0 | 0.0 | 2026-08-16T10:30 |" in history
    watchtower.get_obliterated = lambda: []
    assert "No models obliterated" in ui_watchtower._get_obliteration_history()
    watchtower.get_obliterated = Mock(side_effect=RuntimeError("broken"))
    assert ui_watchtower._get_obliteration_history() == "*Error loading history.*"


def test_frequency_and_auto_queue_handlers_forward_state(monkeypatch):
    callbacks: list[object] = []
    watchtower = SimpleNamespace(
        start_scheduler=Mock(),
        on_new_model=callbacks.append,
        set_status=Mock(),
        _on_new_model_callbacks=callbacks,
    )
    monkeypatch.setattr(ui_watchtower, "_get_watchtower", lambda: watchtower)

    assert "15 minutes" in ui_watchtower._set_scan_frequency("15 minutes")
    watchtower.start_scheduler.assert_called_once_with(interval=900)
    assert "enabled" in ui_watchtower._toggle_auto_obliterate(True)
    callbacks[0](SimpleNamespace(model_id="org/new"))
    watchtower.set_status.assert_called_once_with("org/new", "queued")
    assert "disabled" in ui_watchtower._toggle_auto_obliterate(False)
    assert callbacks == []


def test_refresh_dropdown_handles_an_empty_choice_set(monkeypatch):
    monkeypatch.setattr(ui_watchtower, "_get_trending_choices", lambda: [])

    update = ui_watchtower._refresh_one_click_dropdown()

    assert update["choices"] == []
    assert update["value"] == ""


def test_tab_construction_requires_the_optional_ui_runtime(monkeypatch):
    monkeypatch.setattr(ui_watchtower, "gr", None)

    with pytest.raises(ImportError, match="spaces extra"):
        ui_watchtower.build_watchtower_tabs()