129e4ea1fc
New router intelligence modules (26 files): alert_ingest/store, audit_store, architecture_pressure, backlog_generator/store, cost_analyzer, data_governance, dependency_scanner, drift_analyzer, incident_* (5 files), llm_enrichment, platform_priority_digest, provider_budget, release_check_runner, risk_* (6 files), signature_state_store, sofiia_auto_router, tool_governance New services: - sofiia-console: Dockerfile, adapters/, monitor/nodes/ops/voice modules, launchd, react static - memory-service: integration_endpoints, integrations, voice_endpoints, static UI - aurora-service: full app suite (analysis, job_store, orchestrator, reporting, schemas, subagents) - sofiia-supervisor: new supervisor service - aistalk-bridge-lite: Telegram bridge lite - calendar-service: CalDAV calendar service with reminders - mlx-stt-service / mlx-tts-service: Apple Silicon speech services - binance-bot-monitor: market monitor service - node-worker: STT/TTS memory providers New tools (9): agent_email, browser_tool, contract_tool, observability_tool, oncall_tool, pr_reviewer_tool, repo_tool, safe_code_executor, secure_vault New crews: agromatrix_crew (10 modules: depth_classifier, doc_facts, doc_focus, farm_state, light_reply, llm_factory, memory_manager, proactivity, reflection_engine, session_context, style_adapter, telemetry) Tests: 85+ test files for all new modules Made-with: Cursor
379 lines
18 KiB
Python
379 lines
18 KiB
Python
"""
|
||
tests/test_stepan_v4_vision_guard.py
|
||
|
||
Unit tests for Vision Consistency Guard (v4.0.1).
|
||
|
||
Scope: vision_guard.py — без залежностей від crewai, httpx, memory-service.
|
||
"""
|
||
from __future__ import annotations
|
||
import time
|
||
import pytest
|
||
|
||
import vision_guard as vg
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def clear_lock():
|
||
"""Очищуємо глобальний стор перед кожним тестом."""
|
||
vg._VISION_LOCK.clear()
|
||
yield
|
||
vg._VISION_LOCK.clear()
|
||
|
||
|
||
# ─── _photo_key ───────────────────────────────────────────────────────────────
|
||
|
||
class TestPhotoKey:
|
||
def test_prefers_file_unique_id(self):
|
||
assert vg._photo_key("file_id_1", "unique_abc") == "unique_abc"
|
||
|
||
def test_fallback_to_file_id(self):
|
||
assert vg._photo_key("file_id_1", None) == "file_id_1"
|
||
|
||
def test_fallback_empty_unique(self):
|
||
assert vg._photo_key("file_id_1", "") == "file_id_1"
|
||
|
||
def test_strips_whitespace(self):
|
||
assert vg._photo_key("fid", " uid ") == "uid"
|
||
|
||
|
||
# ─── extract_label_from_response ─────────────────────────────────────────────
|
||
|
||
class TestExtractLabel:
|
||
def test_crop_corn(self):
|
||
label, conf = vg.extract_label_from_response(
|
||
"На фото кукурудза на стадії V4. Листки виглядають здоровими."
|
||
)
|
||
assert label == "кукурудза"
|
||
assert conf == "high"
|
||
|
||
def test_crop_wheat(self):
|
||
label, conf = vg.extract_label_from_response("Це пшениця, стан задовільний.")
|
||
assert label == "пшениця"
|
||
|
||
def test_crop_sunflower_en(self):
|
||
label, conf = vg.extract_label_from_response("The photo shows a sunflower field.")
|
||
assert label == "sunflower"
|
||
assert conf == "high"
|
||
|
||
def test_diagnosis(self):
|
||
label, conf = vg.extract_label_from_response(
|
||
"Помітний хлороз міжжилкового типу — можливий дефіцит мангану або заліза."
|
||
)
|
||
assert "хлороз" in label or "дефіцит" in label
|
||
|
||
def test_low_confidence_flag(self):
|
||
_, conf = vg.extract_label_from_response(
|
||
"Можливо, це кукурудза, але важко сказати без деталей."
|
||
)
|
||
assert conf == "low"
|
||
|
||
def test_high_confidence_flag(self):
|
||
_, conf = vg.extract_label_from_response("Це кукурудза.")
|
||
assert conf == "high"
|
||
|
||
def test_unknown_when_no_label(self):
|
||
label, conf = vg.extract_label_from_response("Гарне фото поля.")
|
||
assert label == ""
|
||
assert conf == "unknown"
|
||
|
||
def test_empty_string(self):
|
||
label, conf = vg.extract_label_from_response("")
|
||
assert label == ""
|
||
assert conf == "unknown"
|
||
|
||
|
||
# ─── set_vision_lock / get_vision_lock ───────────────────────────────────────
|
||
|
||
class TestVisionLock:
|
||
def test_set_and_get(self):
|
||
vg.set_vision_lock("agromatrix", "chat1", "fid_1", "кукурудза", "high",
|
||
file_unique_id="uid_1")
|
||
lock = vg.get_vision_lock("agromatrix", "chat1")
|
||
assert lock["photo_key"] == "uid_1"
|
||
assert lock["file_id"] == "fid_1"
|
||
assert lock["label"] == "кукурудза"
|
||
assert lock["confidence"] == "high"
|
||
|
||
def test_photo_key_fallback_to_file_id(self):
|
||
vg.set_vision_lock("agromatrix", "chat1", "fid_1", "пшениця", "high")
|
||
lock = vg.get_vision_lock("agromatrix", "chat1")
|
||
assert lock["photo_key"] == "fid_1"
|
||
|
||
def test_different_chats_isolated(self):
|
||
vg.set_vision_lock("agromatrix", "chat1", "f1", "пшениця", "high", file_unique_id="u1")
|
||
vg.set_vision_lock("agromatrix", "chat2", "f2", "соняшник", "low", file_unique_id="u2")
|
||
assert vg.get_vision_lock("agromatrix", "chat1")["label"] == "пшениця"
|
||
assert vg.get_vision_lock("agromatrix", "chat2")["label"] == "соняшник"
|
||
|
||
def test_empty_before_set(self):
|
||
assert vg.get_vision_lock("agromatrix", "chat99") == {}
|
||
|
||
def test_ttl_expiry(self):
|
||
vg.set_vision_lock("agromatrix", "chatX", "fid", "ріпак", "high")
|
||
key = "agromatrix:chatX"
|
||
vg._VISION_LOCK[key]["ts"] = time.time() - vg.VISION_LOCK_TTL - 1
|
||
assert vg.get_vision_lock("agromatrix", "chatX") == {}
|
||
|
||
def test_preserves_user_label_on_update(self):
|
||
vg.set_vision_lock("agromatrix", "chat1", "fid_1", "кукурудза", "high",
|
||
file_unique_id="uid_1")
|
||
vg.set_user_label("agromatrix", "chat1", "Соняшник")
|
||
# Оновлюємо lock — user_label має зберегтися
|
||
vg.set_vision_lock("agromatrix", "chat1", "fid_2", "пшениця", "high",
|
||
file_unique_id="uid_2")
|
||
lock = vg.get_vision_lock("agromatrix", "chat1")
|
||
assert lock["user_label"] == "соняшник"
|
||
|
||
def test_clear_vision_lock(self):
|
||
vg.set_vision_lock("agromatrix", "chat1", "fid", "кукурудза", "high")
|
||
vg.clear_vision_lock("agromatrix", "chat1")
|
||
assert vg.get_vision_lock("agromatrix", "chat1") == {}
|
||
|
||
def test_clear_nonexistent_no_error(self):
|
||
vg.clear_vision_lock("agromatrix", "no_such_chat") # не кидає
|
||
|
||
|
||
# ─── set_user_label / detect_user_override ───────────────────────────────────
|
||
|
||
class TestUserOverride:
|
||
# ── Позитивні кейси ──────────────────────────────────────────────────────
|
||
def test_detect_plain(self):
|
||
assert vg.detect_user_override("це соняшник") == "соняшник"
|
||
|
||
def test_detect_with_punctuation(self):
|
||
assert vg.detect_user_override("це пшениця!") == "пшениця"
|
||
|
||
def test_detect_word_only(self):
|
||
result = vg.detect_user_override("кукурудза")
|
||
assert result == "кукурудза"
|
||
|
||
def test_detect_bur_yan(self):
|
||
result = vg.detect_user_override("це бур'ян")
|
||
assert result != ""
|
||
|
||
def test_detect_grunt(self):
|
||
result = vg.detect_user_override("це ґрунт")
|
||
assert result != ""
|
||
|
||
def test_detect_shkidnyk(self):
|
||
result = vg.detect_user_override("це шкідник")
|
||
assert result != ""
|
||
|
||
# ── Негація — заборона ───────────────────────────────────────────────────
|
||
def test_negation_ce_ne(self):
|
||
assert vg.detect_user_override("це не соняшник") == ""
|
||
|
||
def test_negation_to_ne(self):
|
||
assert vg.detect_user_override("то не пшениця") == ""
|
||
|
||
def test_negation_just_ne(self):
|
||
assert vg.detect_user_override("не кукурудза") == ""
|
||
|
||
# ── Не override ──────────────────────────────────────────────────────────
|
||
def test_no_override_long_text(self):
|
||
assert vg.detect_user_override("покажи скільки там добрив") == ""
|
||
|
||
def test_no_override_question(self):
|
||
assert vg.detect_user_override("що це за хвороба?") == ""
|
||
|
||
def test_no_override_not_in_whitelist(self):
|
||
assert vg.detect_user_override("це якийсь невідомий об'єкт") == ""
|
||
|
||
# ── set_user_label ────────────────────────────────────────────────────────
|
||
def test_set_and_read_user_label(self):
|
||
vg.set_vision_lock("agromatrix", "c1", "f1", "кукурудза", "high")
|
||
vg.set_user_label("agromatrix", "c1", "Соняшник")
|
||
lock = vg.get_vision_lock("agromatrix", "c1")
|
||
assert lock["user_label"] == "соняшник"
|
||
|
||
def test_set_user_label_no_lock(self):
|
||
"""set_user_label без попереднього lock — не має падати."""
|
||
vg.set_user_label("agromatrix", "newchat", "ячмінь")
|
||
lock = vg.get_vision_lock("agromatrix", "newchat")
|
||
assert lock.get("user_label") == "ячмінь"
|
||
|
||
|
||
# ─── is_reeval_request ───────────────────────────────────────────────────────
|
||
|
||
class TestReeval:
|
||
def test_reeval_explicit(self):
|
||
assert vg.is_reeval_request("переоцінити це фото")
|
||
assert vg.is_reeval_request("перевір ще раз")
|
||
assert vg.is_reeval_request("не те, переглянь")
|
||
|
||
def test_no_reeval(self):
|
||
assert not vg.is_reeval_request("що з цим фото?")
|
||
assert not vg.is_reeval_request("дякую")
|
||
assert not vg.is_reeval_request("")
|
||
|
||
|
||
# ─── should_skip_reanalysis ──────────────────────────────────────────────────
|
||
|
||
class TestShouldSkip:
|
||
def test_skip_same_file_unique_id(self):
|
||
vg.set_vision_lock("agromatrix", "c1", "fid_1", "кукурудза", "high",
|
||
file_unique_id="uid_same")
|
||
assert vg.should_skip_reanalysis("agromatrix", "c1", "fid_1", "що там?",
|
||
file_unique_id="uid_same")
|
||
|
||
def test_skip_same_file_different_sizes(self):
|
||
"""Різні file_id але той самий file_unique_id (різні розміри) → skip."""
|
||
vg.set_vision_lock("agromatrix", "c1", "fid_small", "пшениця", "high",
|
||
file_unique_id="uid_photo_1")
|
||
# Надсилається той самий file_unique_id але з іншим file_id (великий розмір)
|
||
assert vg.should_skip_reanalysis("agromatrix", "c1", "fid_large", "що там?",
|
||
file_unique_id="uid_photo_1")
|
||
|
||
def test_no_skip_different_unique_id(self):
|
||
vg.set_vision_lock("agromatrix", "c1", "fid_1", "кукурудза", "high",
|
||
file_unique_id="uid_1")
|
||
assert not vg.should_skip_reanalysis("agromatrix", "c1", "fid_2", "що там?",
|
||
file_unique_id="uid_2")
|
||
|
||
def test_no_skip_different_file_id_no_unique(self):
|
||
vg.set_vision_lock("agromatrix", "c1", "fid_old", "кукурудза", "high")
|
||
assert not vg.should_skip_reanalysis("agromatrix", "c1", "fid_new", "що там?")
|
||
|
||
def test_reeval_clears_lock(self):
|
||
"""Rule C: reeval_request → clear_lock → skip=False."""
|
||
vg.set_vision_lock("agromatrix", "c1", "fid_same", "кукурудза", "high",
|
||
file_unique_id="uid_same")
|
||
result = vg.should_skip_reanalysis("agromatrix", "c1", "fid_same",
|
||
"перевір ще раз",
|
||
file_unique_id="uid_same")
|
||
assert result is False
|
||
# Lock має бути очищено
|
||
assert vg.get_vision_lock("agromatrix", "c1") == {}
|
||
|
||
def test_no_skip_no_lock(self):
|
||
assert not vg.should_skip_reanalysis("agromatrix", "c_empty", "fid_x", "")
|
||
|
||
def test_no_skip_expired_lock(self):
|
||
vg.set_vision_lock("agromatrix", "c1", "fid_same", "пшениця", "high",
|
||
file_unique_id="uid_same")
|
||
vg._VISION_LOCK["agromatrix:c1"]["ts"] = time.time() - vg.VISION_LOCK_TTL - 1
|
||
assert not vg.should_skip_reanalysis("agromatrix", "c1", "fid_same", "",
|
||
file_unique_id="uid_same")
|
||
|
||
|
||
# ─── build_low_confidence_clarifier ──────────────────────────────────────────
|
||
|
||
class TestLowConfidenceClarifier:
|
||
def test_appends_clarifier_when_low_conf(self):
|
||
text = "Можливо, це кукурудза, але важко сказати без деталей."
|
||
result, added = vg.build_low_confidence_clarifier(text)
|
||
assert added is True
|
||
assert len(result) > len(text)
|
||
assert "?" in result
|
||
|
||
def test_no_change_when_high_conf(self):
|
||
text = "Це кукурудза на стадії V4."
|
||
result, added = vg.build_low_confidence_clarifier(text)
|
||
assert added is False
|
||
assert result == text
|
||
|
||
def test_no_duplicate_clarifier(self):
|
||
"""Якщо відповідь вже містить '?' наприкінці — не дублювати."""
|
||
text = "Можливо, це пшениця. Хочете уточнити?"
|
||
result, added = vg.build_low_confidence_clarifier(text)
|
||
assert added is False # вже є '?'
|
||
|
||
def test_empty_input(self):
|
||
result, added = vg.build_low_confidence_clarifier("")
|
||
assert result == ""
|
||
assert added is False
|
||
|
||
def test_returns_tuple(self):
|
||
out = vg.build_low_confidence_clarifier("Це пшениця.")
|
||
assert isinstance(out, tuple) and len(out) == 2
|
||
|
||
|
||
# ─── build_locked_reply ───────────────────────────────────────────────────────
|
||
|
||
class TestLockedReply:
|
||
def test_high_conf(self):
|
||
lock = {"file_id": "f1", "photo_key": "u1", "label": "кукурудза",
|
||
"confidence": "high", "user_label": ""}
|
||
reply = vg.build_locked_reply(lock, "що там?")
|
||
assert "кукурудза" in reply.lower() or "Кукурудза" in reply
|
||
assert "?" in reply
|
||
|
||
def test_user_label_takes_priority(self):
|
||
lock = {"file_id": "f1", "photo_key": "u1", "label": "пшениця",
|
||
"confidence": "high", "user_label": "соняшник"}
|
||
reply = vg.build_locked_reply(lock, "")
|
||
assert "соняшник" in reply.lower() or "Соняшник" in reply
|
||
assert "підтвердив" in reply
|
||
|
||
def test_low_conf(self):
|
||
lock = {"file_id": "f1", "photo_key": "u1", "label": "ріпак",
|
||
"confidence": "low", "user_label": ""}
|
||
reply = vg.build_locked_reply(lock, "")
|
||
assert "ріпак" in reply.lower() or "Ріпак" in reply
|
||
assert "впевненість" in reply.lower() or "переоцінити" in reply.lower()
|
||
|
||
def test_empty_label(self):
|
||
lock = {"file_id": "f1", "photo_key": "u1", "label": "",
|
||
"confidence": "unknown", "user_label": ""}
|
||
reply = vg.build_locked_reply(lock, "")
|
||
assert len(reply) > 0 # fallback рядок
|
||
|
||
|
||
# ─── TTL та cleanup ───────────────────────────────────────────────────────────
|
||
|
||
class TestTTL:
|
||
def test_cleanup_removes_expired(self):
|
||
vg.set_vision_lock("agromatrix", "c_exp", "f1", "ячмінь", "high",
|
||
file_unique_id="u_exp")
|
||
vg._VISION_LOCK["agromatrix:c_exp"]["ts"] = time.time() - vg.VISION_LOCK_TTL - 5
|
||
vg.set_vision_lock("agromatrix", "c_fresh", "f2", "горох", "high",
|
||
file_unique_id="u_fresh")
|
||
vg._cleanup()
|
||
assert "agromatrix:c_exp" not in vg._VISION_LOCK
|
||
assert "agromatrix:c_fresh" in vg._VISION_LOCK
|
||
|
||
def test_lock_ttl_constant_correct(self):
|
||
assert vg.VISION_LOCK_TTL == 1800.0
|
||
|
||
|
||
# ─── Smoke-test сценарій: повний цикл ────────────────────────────────────────
|
||
|
||
class TestSmokeCycle:
|
||
def test_full_cycle(self):
|
||
"""
|
||
Сценарій:
|
||
1. Надіслали фото → lock_set (кукурудза, high)
|
||
2. Повторно той самий file_unique_id → skip
|
||
3. Юзер пише "це соняшник" → user_label
|
||
4. Repeat → locked_reply з user_label
|
||
5. "перевір ще раз" → clear_lock → no skip
|
||
"""
|
||
# 1. Lock після аналізу
|
||
vg.set_vision_lock("agromatrix", "chat_smoke", "fid_a", "кукурудза", "high",
|
||
file_unique_id="uid_photo")
|
||
lock = vg.get_vision_lock("agromatrix", "chat_smoke")
|
||
assert lock["label"] == "кукурудза"
|
||
|
||
# 2. Той самий file_unique_id, інший file_id (розмір) → skip
|
||
skip = vg.should_skip_reanalysis("agromatrix", "chat_smoke",
|
||
"fid_b", "що там?",
|
||
file_unique_id="uid_photo")
|
||
assert skip is True
|
||
|
||
# 3. User override "це соняшник"
|
||
override = vg.detect_user_override("це соняшник")
|
||
assert override == "соняшник"
|
||
vg.set_user_label("agromatrix", "chat_smoke", override)
|
||
|
||
# 4. Locked reply — має повернути соняшник (user_label)
|
||
lock2 = vg.get_vision_lock("agromatrix", "chat_smoke")
|
||
reply = vg.build_locked_reply(lock2, "знову надіслав фото")
|
||
assert "соняшник" in reply.lower() or "Соняшник" in reply
|
||
|
||
# 5. Reeval → clear_lock → no skip
|
||
skip2 = vg.should_skip_reanalysis("agromatrix", "chat_smoke",
|
||
"fid_b", "перевір ще раз",
|
||
file_unique_id="uid_photo")
|
||
assert skip2 is False
|
||
assert vg.get_vision_lock("agromatrix", "chat_smoke") == {}
|