J.A.R.V.I.S-py/tests/test_wave59_handlers.py

"""Tests for `wave59_handlers.py` — Python parity for the Rust-only
wol / banter / conversation / ddg_answer packs.

Focus is on the testable pure logic — MAC normalisation, magic-packet
construction, trigger stripping, banter state machine. Network calls
(DDG, LLM) are skipped here; they're integration concerns covered by
the Rust pack tests.
"""

import importlib
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import wave59_handlers


def _rewire(isolated_state, *modules):
    importlib.reload(wave59_handlers)
    out = []
    for name in modules:
        mod = isolated_state(name, ['_PATH', '_PROFILES_DIR'])
        setattr(wave59_handlers, name, mod)
        out.append(mod)
    return out


def _capture():
    captured = []
    wave59_handlers.set_speak(captured.append)
    return captured


# ── WOL ─────────────────────────────────────────────────────────────────


def test_normalise_mac_accepts_common_formats():
    n = wave59_handlers._normalise_mac
    assert n("AA:BB:CC:DD:EE:FF") == "AABBCCDDEEFF"
    assert n("aa-bb-cc-dd-ee-ff") == "AABBCCDDEEFF"
    assert n("AABB.CCDD.EEFF") == "AABBCCDDEEFF"
    assert n("AABBCCDDEEFF") == "AABBCCDDEEFF"
    assert n("garbage") is None
    assert n("") is None
    assert n("AABBCCDD") is None  # too short


def test_build_magic_packet_shape():
    p = wave59_handlers._build_magic_packet("AABBCCDDEEFF")
    # 6 × 0xFF + 16 × MAC = 102 bytes
    assert len(p) == 102
    assert p[:6] == b"\xff" * 6
    # First repeat of MAC starts at byte 6
    assert p[6:12] == bytes.fromhex("AABBCCDDEEFF")
    # Last repeat at offset 96
    assert p[96:102] == bytes.fromhex("AABBCCDDEEFF")


def test_wol_speaks_help_when_mac_not_set(isolated_state, monkeypatch):
    (mem,) = _rewire(isolated_state, 'memory_store')
    monkeypatch.delenv("JARVIS_WOL_TARGET", raising=False)
    spoken = _capture()
    wave59_handlers.do_wol()
    assert "MAC" in spoken[0]


def test_wol_speaks_malformed_when_mac_invalid(isolated_state, monkeypatch):
    (mem,) = _rewire(isolated_state, 'memory_store')
    monkeypatch.delenv("JARVIS_WOL_TARGET", raising=False)
    mem.remember("wol_server", "totally-not-a-mac")
    spoken = _capture()
    wave59_handlers.do_wol()
    assert "подозрительно" in spoken[0]


# ── banter ──────────────────────────────────────────────────────────────


def test_banter_pause_then_fire_says_quiet(isolated_state):
    importlib.reload(wave59_handlers)
    spoken = _capture()
    wave59_handlers.do_banter_pause()
    wave59_handlers.do_banter_fire()
    # Two utterances captured: "Молчу..." then a "без комментариев" reply
    assert any("Молчу" in s for s in spoken)
    assert any("без комментариев" in s.lower() for s in spoken)


def test_banter_resume_clears_pause(isolated_state):
    importlib.reload(wave59_handlers)
    spoken = _capture()
    wave59_handlers.do_banter_pause()
    wave59_handlers.do_banter_resume()
    wave59_handlers.do_banter_fire()
    # After resume, fire should return one of the known pool lines.
    assert any(s in wave59_handlers._BANTER_LINES for s in spoken)


# ── conversation ────────────────────────────────────────────────────────


def test_conversation_repeat_with_no_history(isolated_state):
    importlib.reload(wave59_handlers)
    wave59_handlers.set_history_ref(None)
    spoken = _capture()
    wave59_handlers.do_conversation_repeat()
    assert "ещё ничего не говорил" in spoken[0]


def test_conversation_repeat_speaks_last_assistant(isolated_state):
    importlib.reload(wave59_handlers)
    history = [
        {"role": "system", "content": "you are jarvis"},
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "Слушаю, сэр."},
        {"role": "user", "content": "what time"},
        {"role": "assistant", "content": "Восемь утра, сэр."},
    ]
    wave59_handlers.set_history_ref(history)
    spoken = _capture()
    wave59_handlers.do_conversation_repeat()
    assert spoken[0] == "Восемь утра, сэр."


def test_conversation_summary_empty_history(isolated_state):
    importlib.reload(wave59_handlers)
    wave59_handlers.set_history_ref([{"role": "system", "content": "x"}])
    spoken = _capture()
    wave59_handlers.do_conversation_summary()
    assert "пока ещё ни о чём" in spoken[0]


# ── DDG trigger stripping ───────────────────────────────────────────────


def test_strip_ddg_trigger_known_prefixes():
    s = wave59_handlers._strip_ddg_trigger
    assert s("что такое квантовая запутанность") == "квантовая запутанность"
    assert s("кто такой никола тесла") == "никола тесла"
    assert s("what is the speed of light") == "the speed of light"
    assert s("tell me about jupiter") == "jupiter"
    # No trigger → phrase passes through unchanged
    assert s("просто текст без триггера") == "просто текст без триггера"
    assert s("") == ""


def test_ddg_answer_with_empty_query_says_so(isolated_state):
    importlib.reload(wave59_handlers)
    spoken = _capture()
    wave59_handlers.do_ddg_answer(voice="что такое")
    assert "Что искать" in spoken[0]