microdao-daarion/tests/test_stepan_doc_focus.py

"""
Tests for Doc Focus Gate (v3.5):
- PROMPT A: doc_focus fields in session, is_doc_focus_active, TTL
- PROMPT B: _is_doc_question, context_mode arbitration
- PROMPT C: auto-clear on vision/URL domain
- PROMPT D: /doc on|off|status operator commands
- PROMPT 1: domain override — explicit doc-token beats vision
- PROMPT 2: TTL auto-expire (simulated)
- PROMPT 3: context bleed guard
"""
import sys
import os
import time

sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "gateway-bot"))


# ── PROMPT A: session_context ────────────────────────────────────────────────

def test_default_session_has_doc_focus():
    from crews.agromatrix_crew.session_context import _default_session
    s = _default_session()
    assert "doc_focus" in s
    assert s["doc_focus"] is False
    assert "doc_focus_ts" in s
    assert s["doc_focus_ts"] == 0.0


def test_is_doc_focus_active_false_by_default():
    from crews.agromatrix_crew.session_context import _default_session, is_doc_focus_active
    s = _default_session()
    assert is_doc_focus_active(s) is False


def test_is_doc_focus_active_true_when_set():
    from crews.agromatrix_crew.session_context import is_doc_focus_active
    now = time.time()
    s = {"doc_focus": True, "doc_focus_ts": now}
    assert is_doc_focus_active(s, now) is True


def test_is_doc_focus_active_expired():
    from crews.agromatrix_crew.session_context import is_doc_focus_active, DOC_FOCUS_TTL
    old_ts = time.time() - DOC_FOCUS_TTL - 1  # протух
    s = {"doc_focus": True, "doc_focus_ts": old_ts}
    assert is_doc_focus_active(s) is False


def test_is_doc_focus_active_just_within_ttl():
    from crews.agromatrix_crew.session_context import is_doc_focus_active, DOC_FOCUS_TTL
    now = time.time()
    recent_ts = now - DOC_FOCUS_TTL + 5  # 5 секунд до завершення
    s = {"doc_focus": True, "doc_focus_ts": recent_ts}
    assert is_doc_focus_active(s, now) is True


def test_update_session_doc_focus():
    from crews.agromatrix_crew.session_context import update_session, load_session, clear_session
    chat_id = "test_doc_focus_chat"
    clear_session(chat_id)
    now = time.time()
    update_session(chat_id, "test", depth="deep", doc_focus=True, doc_focus_ts=now)
    s = load_session(chat_id)
    assert s["doc_focus"] is True
    assert abs(s["doc_focus_ts"] - now) < 1.0
    clear_session(chat_id)


def test_update_session_doc_focus_off():
    from crews.agromatrix_crew.session_context import update_session, load_session, clear_session
    chat_id = "test_doc_focus_off_chat"
    clear_session(chat_id)
    now = time.time()
    # Спочатку вмикаємо
    update_session(chat_id, "test", depth="deep", doc_focus=True, doc_focus_ts=now)
    # Потім вимикаємо
    update_session(chat_id, "test2", depth="deep", doc_focus=False, doc_focus_ts=0.0)
    s = load_session(chat_id)
    assert s["doc_focus"] is False
    clear_session(chat_id)


def test_is_doc_focus_active_fail_safe():
    from crews.agromatrix_crew.session_context import is_doc_focus_active
    # Broken session
    assert is_doc_focus_active(None) is False
    assert is_doc_focus_active({}) is False
    assert is_doc_focus_active({"doc_focus": True}) is False  # немає ts → вважається протухлим


# ── PROMPT B: _is_doc_question ───────────────────────────────────────────────

def test_is_doc_question_explicit():
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("що у звіті?") is True
    assert _is_doc_question("прибуток у документі") is True
    assert _is_doc_question("відкрий звіт") is True
    assert _is_doc_question("в цьому документі є дані?") is True


def test_is_doc_question_financial_with_doc_anchor():
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("прибуток зі звіту") is True
    assert _is_doc_question("витрати у файлі xlsx") is True
    assert _is_doc_question("скільки грн/га в документі") is True


def test_is_doc_question_url_returns_false():
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("https://www.cropscience.bayer.ua/Products") is False
    assert _is_doc_question("вивчи каталог https://bayer.com") is False


def test_is_doc_question_vision_returns_false():
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("що з листям?") is False
    assert _is_doc_question("плями на рослині — що це?") is False
    assert _is_doc_question("шкідник на фото") is False


def test_is_doc_question_web_intent_returns_false():
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("вивчи каталог засобів захисту") is False
    assert _is_doc_question("переглянь сторінку сайту") is False


def test_is_doc_question_general_false():
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("привіт") is False
    assert _is_doc_question("що ти вмієш?") is False
    assert _is_doc_question("план на тиждень") is False


# ── PROMPT B: _detect_domain ─────────────────────────────────────────────────

def test_detect_domain_vision():
    from crews.agromatrix_crew.doc_focus import _detect_domain
    assert _detect_domain("що з листям?") == "vision"
    assert _detect_domain("плями на рослині") == "vision"
    assert _detect_domain("хвороба кукурудзи") == "vision"


def test_detect_domain_web():
    from crews.agromatrix_crew.doc_focus import _detect_domain
    assert _detect_domain("https://bayer.com") == "web"
    assert _detect_domain("вивчи каталог сайту") == "web"


def test_detect_domain_doc():
    from crews.agromatrix_crew.doc_focus import _detect_domain
    assert _detect_domain("що у звіті?") == "doc"
    assert _detect_domain("прибуток у документі") == "doc"


def test_detect_domain_general():
    from crews.agromatrix_crew.doc_focus import _detect_domain
    assert _detect_domain("привіт") == "general"
    assert _detect_domain("як справи?") == "general"


# ── PROMPT C: auto-clear logic ────────────────────────────────────────────────

def test_doc_focus_clear_on_vision():
    """Симулюємо логіку auto-clear: domain=vision → doc_focus=False."""
    from crews.agromatrix_crew.doc_focus import _detect_domain
    from crews.agromatrix_crew.session_context import is_doc_focus_active

    now = time.time()
    session = {"doc_focus": True, "doc_focus_ts": now}
    text = "що з листям на фото?"
    domain = _detect_domain(text)

    assert domain == "vision"
    # Логіка з run.py:
    if domain in ("vision", "web"):
        session["doc_focus"] = False
        session["doc_focus_ts"] = 0.0

    assert is_doc_focus_active(session, now) is False


def test_doc_focus_clear_on_url():
    """URL message → doc_focus скидається."""
    from crews.agromatrix_crew.doc_focus import _detect_domain
    from crews.agromatrix_crew.session_context import is_doc_focus_active

    now = time.time()
    session = {"doc_focus": True, "doc_focus_ts": now}
    text = "ще будь ласка вивчи каталог засобів захисту рослин фірми BAYER https://www.cropscience.bayer.ua"
    domain = _detect_domain(text)

    assert domain == "web"
    if domain in ("vision", "web"):
        session["doc_focus"] = False
        session["doc_focus_ts"] = 0.0

    assert is_doc_focus_active(session, now) is False


def test_doc_focus_preserved_on_doc_question():
    """Doc питання → doc_focus не скидається."""
    from crews.agromatrix_crew.doc_focus import _detect_domain, _is_doc_question
    from crews.agromatrix_crew.session_context import is_doc_focus_active

    now = time.time()
    session = {"doc_focus": True, "doc_focus_ts": now}
    text = "який прибуток у звіті?"

    domain = _detect_domain(text)
    assert domain == "doc"
    # Vision/web auto-clear не спрацьовує
    assert domain not in ("vision", "web")
    # doc_focus залишається
    assert is_doc_focus_active(session, now) is True


# ── PROMPT D: /doc commands ───────────────────────────────────────────────────

def test_doc_command_status_no_session():
    from crews.agromatrix_crew.session_context import clear_session
    from crews.agromatrix_crew.doc_focus import handle_doc_focus
    clear_session("doc_test_chat_99")
    result = handle_doc_focus("status", chat_id="doc_test_chat_99")
    assert result is not None
    msg = result.get("message", "") or str(result)
    assert "doc_focus=off" in msg or "off" in msg


def test_doc_command_on():
    from crews.agromatrix_crew.session_context import clear_session, load_session
    from crews.agromatrix_crew.doc_focus import handle_doc_focus
    chat_id = "doc_on_test_chat"
    clear_session(chat_id)
    result = handle_doc_focus("on", chat_id=chat_id)
    assert result is not None
    msg = result.get("message", "") or str(result)
    assert "on" in msg
    # Сесія має відображати зміну
    from crews.agromatrix_crew.session_context import _STORE
    s = _STORE.get(chat_id) or {}
    assert s.get("doc_focus") is True


def test_doc_command_off():
    from crews.agromatrix_crew.session_context import _STORE, clear_session
    from crews.agromatrix_crew.doc_focus import handle_doc_focus
    chat_id = "doc_off_test_chat"
    _STORE[chat_id] = {"doc_focus": True, "doc_focus_ts": time.time(), "updated_at": time.time()}
    result = handle_doc_focus("off", chat_id=chat_id)
    assert result is not None
    msg = result.get("message", "") or str(result)
    assert "off" in msg
    s = _STORE.get(chat_id) or {}
    assert s.get("doc_focus") is False
    clear_session(chat_id)


def test_doc_command_no_chat_id():
    from crews.agromatrix_crew.doc_focus import handle_doc_focus
    result = handle_doc_focus("on", chat_id=None)
    msg = result.get("message", "") or str(result)
    assert "chat_id" in msg.lower() or "required" in msg.lower()


def test_doc_command_status_shows_ttl():
    from crews.agromatrix_crew.session_context import _STORE, clear_session
    from crews.agromatrix_crew.doc_focus import handle_doc_focus
    chat_id = "doc_status_ttl_test"
    now = time.time()
    _STORE[chat_id] = {"doc_focus": True, "doc_focus_ts": now, "active_doc_id": "abc123",
                       "doc_facts": {"profit_uah": 5000000}, "updated_at": now}
    result = handle_doc_focus("status", chat_id=chat_id)
    msg = result.get("message", "") or str(result)
    assert "on" in msg
    assert "ttl_left" in msg
    assert "abc123" in msg
    assert "profit_uah" in msg
    clear_session(chat_id)


# ── PROMPT 1: domain override ────────────────────────────────────────────────

def test_detect_domain_explicit_doc_beats_vision():
    """Explicit doc-токен ('по звіту', 'у файлі') перемагає vision навіть якщо є рослини/листя."""
    from crews.agromatrix_crew.doc_focus import _detect_domain
    # Скрін таблиці + питання по звіту
    assert _detect_domain("по звіту: що з листям?") == "doc"
    assert _detect_domain("у файлі є дані по хворобі рослин?") == "doc"
    assert _detect_domain("в документі — плями чи хвороба?") == "doc"


def test_detect_domain_empty_text_is_vision():
    """Порожній текст (caption відсутній) → vision (фото без caption)."""
    from crews.agromatrix_crew.doc_focus import _detect_domain
    assert _detect_domain("") == "vision"
    assert _detect_domain("   ") == "vision"


def test_detect_domain_url_beats_explicit_doc():
    """URL завжди виграє навіть якщо є explicit doc-токен."""
    from crews.agromatrix_crew.doc_focus import _detect_domain
    assert _detect_domain("по звіту https://bayer.com/products") == "web"


def test_is_doc_question_explicit_doc_token_beats_vision():
    """_is_doc_question повертає True якщо explicit doc-токен, навіть якщо є vision-слова."""
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("по звіту є дані про шкідника?") is True
    assert _is_doc_question("у файлі листя згадується?") is True
    assert _is_doc_question("в документі хвороба кукурудзи?") is True


def test_is_doc_question_vision_only_without_doc_token():
    """Vision без explicit doc-токена → False."""
    from crews.agromatrix_crew.doc_focus import _is_doc_question
    assert _is_doc_question("що з листям на фото?") is False
    assert _is_doc_question("є хвороба чи шкідник?") is False


# ── PROMPT 2: TTL auto-expire ────────────────────────────────────────────────

def test_ttl_auto_expire_logic():
    """Симулюємо: doc_focus=True але TTL протух → expire."""
    from crews.agromatrix_crew.session_context import is_doc_focus_active, DOC_FOCUS_TTL

    now = time.time()
    old_ts = now - DOC_FOCUS_TTL - 60  # на хвилину старіше TTL

    session = {
        "doc_focus": True,
        "doc_focus_ts": old_ts,
        "active_doc_id": "some_doc",
    }

    # is_doc_focus_active має повернути False
    assert is_doc_focus_active(session, now) is False

    # Логіка expire з run.py:
    expired_age = round(now - (session.get("doc_focus_ts") or 0.0))
    if session.get("doc_focus") and not is_doc_focus_active(session, now):
        session["doc_focus"] = False
        session["doc_focus_ts"] = 0.0

    assert session["doc_focus"] is False
    assert session["doc_focus_ts"] == 0.0
    assert expired_age > DOC_FOCUS_TTL


def test_ttl_not_expired_within_window():
    """doc_focus не скидається якщо TTL ще не минув."""
    from crews.agromatrix_crew.session_context import is_doc_focus_active, DOC_FOCUS_TTL

    now = time.time()
    recent_ts = now - (DOC_FOCUS_TTL / 2)  # половина TTL

    session = {"doc_focus": True, "doc_focus_ts": recent_ts}
    assert is_doc_focus_active(session, now) is True

    # Expire logic не спрацьовує
    if session.get("doc_focus") and not is_doc_focus_active(session, now):
        session["doc_focus"] = False
    assert session["doc_focus"] is True


# ── PROMPT 3: context bleed guard ────────────────────────────────────────────

def _apply_bleed_guard(styled_response: str, context_mode: str) -> str:
    """Копія логіки context bleed guard з run.py для тестів."""
    import re as _re
    if context_mode == "general":
        _BLEED_RE = _re.compile(
            r"у\s+(?:цьому|наданому|даному)\s+документі"
            r"|в\s+(?:цьому|наданому|даному)\s+документі"
            r"|у\s+(?:цьому\s+)?звіті|в\s+(?:цьому\s+)?звіті",
            _re.IGNORECASE | _re.UNICODE,
        )
        if _BLEED_RE.search(styled_response):
            return "Щоб відповісти точно, уточни: це питання про звіт чи про інше?"
    return styled_response


def test_bleed_guard_replaces_doc_phrase_in_general_mode():
    """В general-mode фраза 'у цьому документі' заміняється на нейтральне питання."""
    response = "У цьому документі немає інформації про каталог BAYER."
    result = _apply_bleed_guard(response, "general")
    assert "цьому документі" not in result
    assert "уточни" in result


def test_bleed_guard_replaces_zvit_phrase():
    """В general-mode фраза 'у звіті' також блокується."""
    response = "У звіті немає даних про засоби захисту."
    result = _apply_bleed_guard(response, "general")
    assert "уточни" in result


def test_bleed_guard_no_replacement_in_doc_mode():
    """В doc-mode doc-фрази дозволені."""
    response = "У цьому документі прибуток: 5 972 016 грн."
    result = _apply_bleed_guard(response, "doc")
    assert result == response


def test_bleed_guard_no_replacement_without_doc_phrase():
    """Відповідь без doc-фраз не змінюється в general-mode."""
    response = "Це рослина кукурудзи, виглядає здоровою."
    result = _apply_bleed_guard(response, "general")
    assert result == response


def test_bleed_guard_case_insensitive():
    """Блокування не залежить від регістру."""
    response = "В Цьому Документі відсутня ця інформація."
    result = _apply_bleed_guard(response, "general")
    assert "уточни" in result


def test_bleed_guard_variant_phrases():
    """Різні варіанти фрази блокуються."""
    phrases = [
        "у наданому документі немає",
        "В даному документі відсутні",
        "у цьому звіті є лише",
        "в цьому звіті міститься",
    ]
    for p in phrases:
        result = _apply_bleed_guard(p, "general")
        assert "уточни" in result, f"Not blocked: {p!r}"