feat(p1x): de-hardcode VLM models/endpoints to vlm_config (DGX-ready)
Migre les call-sites VLM serveur vers la configuration centrale pour fonctionner sur DGX (tunnel Ollama 11434), où gemma4:* est absent et le port Docker 11435 est mort. - task_planner, replay_verifier, domain_context, ir_builder, resolve_engine (popup): modele -> vlm_config.get_vlm_model(), defaut 11435 -> 11434 (override GEMMA4_PORT legacy conserve) - resolve_engine (grounding bbox x2): nouvel helper vlm_config.get_bbox_grounding_model() (var dediee RPA_BBOX_GROUNDING_MODEL, fallback RPA_GROUNDING_MODEL puis qwen2.5vl:7b-rpa) -> desambiguise le conflit D5-v3b, bbox_2d + num_ctx 4096 preserves - safety_checks_provider: defaut -> get_vlm_model(), override RPA_SAFETY_CHECKS_LLM_MODEL preserve - ui_detector: default_factory + resolution lazy (corrige aussi un gel a l'import), pas d'appel reseau a l'import - field_extractor: property lazy via vlm_config TDD strict (RED->GREEN), 305 tests verts, tests mockes HTTP (zero dependance DGX reel), aucun alias Ollama. Hors perimetre (arbitrage Dom): client Lea agent_v1/executor.py (gele), chemin V4 observe_reason_act (RPA_REASONING_MODEL), core/config.py defaults. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dom
@@ -398,6 +398,56 @@ class TestWorkflowOutcomeLLM:
|
||||
)
|
||||
assert "10 dossiers" in rapport
|
||||
|
||||
def test_refine_modele_via_vlm_config(self):
|
||||
"""Le payload _llm_refine_summary utilise le modèle résolu par vlm_config."""
|
||||
ctx = get_domain_context("tim_codage")
|
||||
captured = {}
|
||||
|
||||
def fake_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
captured["model"] = (json or {}).get("model")
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"message": {"content": "ok"}}
|
||||
return resp
|
||||
|
||||
fake_requests = MagicMock()
|
||||
fake_requests.post.side_effect = fake_post
|
||||
|
||||
with patch.dict("sys.modules", {"requests": fake_requests}), patch(
|
||||
"agent_v0.server_v1.domain_context.vlm_config.get_vlm_model",
|
||||
return_value="modele-resolu:test",
|
||||
):
|
||||
ctx._llm_refine_summary(
|
||||
template="ok", subs={"workflow_name": "x"}, success=True
|
||||
)
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
|
||||
def test_refine_endpoint_par_defaut_11434(self, monkeypatch):
|
||||
"""Sans GEMMA4_PORT, _llm_refine_summary vise 11434, pas le port mort 11435."""
|
||||
monkeypatch.delenv("GEMMA4_PORT", raising=False)
|
||||
ctx = get_domain_context("tim_codage")
|
||||
captured = {}
|
||||
|
||||
def fake_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"message": {"content": "ok"}}
|
||||
return resp
|
||||
|
||||
fake_requests = MagicMock()
|
||||
fake_requests.post.side_effect = fake_post
|
||||
|
||||
with patch.dict("sys.modules", {"requests": fake_requests}):
|
||||
ctx._llm_refine_summary(
|
||||
template="ok", subs={"workflow_name": "x"}, success=True
|
||||
)
|
||||
|
||||
assert ":11434" in captured["url"]
|
||||
assert ":11435" not in captured["url"]
|
||||
|
||||
def test_llm_refine_network_error_safe(self):
|
||||
"""_llm_refine_summary ne doit jamais lever, même si requests échoue."""
|
||||
ctx = get_domain_context("tim_codage")
|
||||
|
||||
@@ -311,6 +311,26 @@ class TestDataStore:
|
||||
|
||||
class TestFieldExtractor:
|
||||
|
||||
def test_model_default_via_vlm_config(self, monkeypatch):
|
||||
"""Sans modèle explicite, ollama_model est résolu via vlm_config (lazy, pas gemma4 en dur)."""
|
||||
monkeypatch.delenv("RPA_VLM_MODEL", raising=False)
|
||||
monkeypatch.delenv("VLM_MODEL", raising=False)
|
||||
from unittest.mock import patch
|
||||
import core.extraction.field_extractor as fe_mod
|
||||
|
||||
extractor = FieldExtractor()
|
||||
with patch.object(fe_mod.vlm_config, "get_vlm_model", return_value="modele-resolu:test"):
|
||||
assert extractor.ollama_model == "modele-resolu:test"
|
||||
|
||||
def test_explicit_model_preserved(self):
|
||||
"""Un modèle explicite est conservé (résolution non déclenchée)."""
|
||||
from unittest.mock import patch
|
||||
import core.extraction.field_extractor as fe_mod
|
||||
extractor = FieldExtractor(ollama_model="mon-modele:pin")
|
||||
with patch.object(fe_mod.vlm_config, "get_vlm_model",
|
||||
side_effect=AssertionError("ne doit pas être appelé")):
|
||||
assert extractor.ollama_model == "mon-modele:pin"
|
||||
|
||||
def test_extract_file_not_found(self, sample_schema):
|
||||
extractor = FieldExtractor()
|
||||
result = extractor.extract_fields("/tmp/nonexistent.png", sample_schema)
|
||||
|
||||
@@ -263,6 +263,54 @@ class TestVerifySemantic:
|
||||
assert result is not None
|
||||
assert result["verified"] is False
|
||||
|
||||
def test_verify_semantic_modele_via_vlm_config(self, verifier, screenshot_white):
|
||||
"""Le payload _verify_semantic utilise le modèle résolu par vlm_config."""
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
captured["model"] = (json or {}).get("model")
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"message": {"content": "VERDICT: OUI\nRAISON: ok"}}
|
||||
return resp
|
||||
|
||||
with patch(
|
||||
"agent_v0.server_v1.replay_verifier.vlm_config.get_vlm_model",
|
||||
return_value="modele-resolu:test",
|
||||
), patch("requests.post", side_effect=capture_post):
|
||||
verifier._verify_semantic(
|
||||
screenshot_before=screenshot_white,
|
||||
screenshot_after=screenshot_white,
|
||||
expected_result="Le fichier est ouvert",
|
||||
)
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
|
||||
def test_verify_semantic_endpoint_par_defaut_11434(
|
||||
self, verifier, screenshot_white, monkeypatch
|
||||
):
|
||||
"""Sans GEMMA4_PORT, _verify_semantic vise 11434, pas le port mort 11435."""
|
||||
monkeypatch.delenv("GEMMA4_PORT", raising=False)
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"message": {"content": "VERDICT: OUI\nRAISON: ok"}}
|
||||
return resp
|
||||
|
||||
with patch("requests.post", side_effect=capture_post):
|
||||
verifier._verify_semantic(
|
||||
screenshot_before=screenshot_white,
|
||||
screenshot_after=screenshot_white,
|
||||
expected_result="Le fichier est ouvert",
|
||||
)
|
||||
|
||||
assert ":11434" in captured["url"]
|
||||
assert ":11435" not in captured["url"]
|
||||
|
||||
@patch("requests.post")
|
||||
def test_vlm_timeout_retourne_none(self, mock_post, verifier, screenshot_white):
|
||||
"""Timeout VLM → retourne None (fallback gracieux)."""
|
||||
|
||||
@@ -47,7 +47,10 @@ def test_all_three_bbox_sites_have_num_ctx_4096():
|
||||
@pytest.mark.unit
|
||||
def test_locate_popup_button_payload_num_ctx(monkeypatch):
|
||||
"""Test runtime : _locate_popup_button construit un payload avec
|
||||
num_ctx=4096 et model=qwen2.5vl:7b."""
|
||||
num_ctx=4096 et modèle bbox résolu via helper (défaut qwen2.5vl:7b-rpa)."""
|
||||
# D5-v3b : sans env, le défaut bbox est qwen2.5vl:7b-rpa (présent DGX).
|
||||
monkeypatch.delenv("RPA_BBOX_GROUNDING_MODEL", raising=False)
|
||||
monkeypatch.delenv("RPA_GROUNDING_MODEL", raising=False)
|
||||
captured = {}
|
||||
|
||||
def fake_post(url, json=None, timeout=None):
|
||||
@@ -78,8 +81,79 @@ def test_locate_popup_button_payload_num_ctx(monkeypatch):
|
||||
f"_locate_popup_button payload sans num_ctx=4096 : "
|
||||
f"{captured['payload']['options']}"
|
||||
)
|
||||
# Modèle non changé
|
||||
assert captured["payload"]["model"] == "qwen2.5vl:7b"
|
||||
# Modèle bbox résolu via helper : défaut qwen2.5vl:7b-rpa (présent DGX)
|
||||
assert captured["payload"]["model"] == "qwen2.5vl:7b-rpa"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_locate_popup_button_model_via_helper(monkeypatch):
|
||||
"""D5-v3b : _locate_popup_button résout le modèle via get_bbox_grounding_model."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_post(url, json=None, timeout=None):
|
||||
captured["payload"] = json
|
||||
resp = MagicMock()
|
||||
resp.ok = False
|
||||
resp.json.return_value = {"message": {"content": ""}}
|
||||
return resp
|
||||
|
||||
import requests
|
||||
monkeypatch.setattr(requests, "post", fake_post)
|
||||
from agent_v0.server_v1 import resolve_engine as re_module
|
||||
|
||||
with patch(
|
||||
"agent_v0.server_v1.resolve_engine.vlm_config.get_bbox_grounding_model",
|
||||
return_value="bbox-modele:test",
|
||||
):
|
||||
re_module._locate_popup_button(
|
||||
screenshot_b64="fake_b64",
|
||||
button_text="OK",
|
||||
screen_width=1920,
|
||||
screen_height=1080,
|
||||
)
|
||||
|
||||
assert captured["payload"]["model"] == "bbox-modele:test"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_resolve_by_grounding_model_via_helper(monkeypatch, tmp_path):
|
||||
"""D5-v3b : _resolve_by_grounding (Essai 2 Ollama) résout via get_bbox_grounding_model."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
from PIL import Image
|
||||
|
||||
# Screenshot réel minimal
|
||||
shot = tmp_path / "shot.png"
|
||||
Image.new("RGB", (200, 120), (255, 255, 255)).save(shot)
|
||||
|
||||
posts = []
|
||||
|
||||
def fake_post(url, json=None, timeout=None):
|
||||
posts.append({"url": url, "payload": json})
|
||||
resp = MagicMock()
|
||||
resp.ok = False # vLLM KO → bascule Ollama ; Ollama contenu vide → None
|
||||
resp.json.return_value = {"message": {"content": ""}}
|
||||
return resp
|
||||
|
||||
import requests
|
||||
monkeypatch.setattr(requests, "post", fake_post)
|
||||
from agent_v0.server_v1 import resolve_engine as re_module
|
||||
|
||||
with patch(
|
||||
"agent_v0.server_v1.resolve_engine.vlm_config.get_bbox_grounding_model",
|
||||
return_value="bbox-modele:test",
|
||||
):
|
||||
re_module._resolve_by_grounding(
|
||||
screenshot_path=str(shot),
|
||||
target_spec={"by_text": "OK"},
|
||||
screen_width=200,
|
||||
screen_height=120,
|
||||
)
|
||||
|
||||
ollama_posts = [p for p in posts if p["url"].endswith("/api/chat")]
|
||||
assert ollama_posts, "Aucun appel Ollama /api/chat capturé"
|
||||
assert ollama_posts[0]["payload"]["model"] == "bbox-modele:test"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
|
||||
58
tests/unit/test_resolve_engine_observer_vlm.py
Normal file
58
tests/unit/test_resolve_engine_observer_vlm.py
Normal file
@@ -0,0 +1,58 @@
|
||||
"""Tests dé-hardcodage VLM de l'Observer (_pre_analyze_screen_sync).
|
||||
|
||||
Le modèle doit venir de vlm_config et l'endpoint viser 11434 (Ollama/tunnel DGX),
|
||||
pas le port mort 11435 ni le modèle gemma4:e4b absent du DGX.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
_ROOT = str(Path(__file__).resolve().parents[2])
|
||||
if _ROOT not in sys.path:
|
||||
sys.path.insert(0, _ROOT)
|
||||
|
||||
from agent_v0.server_v1.resolve_engine import _pre_analyze_screen_sync
|
||||
|
||||
|
||||
def _fake_resp():
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"message": {"content": "ÉTAT: OK\nBOUTON: aucun\nDÉTAIL: rien"}}
|
||||
return resp
|
||||
|
||||
|
||||
def test_observer_modele_via_vlm_config():
|
||||
"""Le payload Observer utilise le modèle résolu par vlm_config."""
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
captured["model"] = (json or {}).get("model")
|
||||
return _fake_resp()
|
||||
|
||||
with patch(
|
||||
"agent_v0.server_v1.resolve_engine.vlm_config.get_vlm_model",
|
||||
return_value="modele-resolu:test",
|
||||
), patch("requests.post", side_effect=capture_post):
|
||||
_pre_analyze_screen_sync("ZmFrZQ==", "écran prêt", "Fenêtre", 1920, 1080)
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
|
||||
|
||||
def test_observer_endpoint_par_defaut_11434(monkeypatch):
|
||||
"""Sans GEMMA4_PORT, l'Observer vise 11434, pas le port mort 11435."""
|
||||
monkeypatch.delenv("GEMMA4_PORT", raising=False)
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
return _fake_resp()
|
||||
|
||||
with patch("requests.post", side_effect=capture_post):
|
||||
_pre_analyze_screen_sync("ZmFrZQ==", "écran prêt", "Fenêtre", 1920, 1080)
|
||||
|
||||
assert ":11434" in captured["url"]
|
||||
assert ":11435" not in captured["url"]
|
||||
@@ -118,3 +118,50 @@ def test_empty_declarative_with_llm_returns_only_llm():
|
||||
)
|
||||
assert len(payload.checks) == 2
|
||||
assert all(c["source"] == "llm_contextual" for c in payload.checks)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Dé-hardcodage VLM (P1.x) : modèle safety-checks résolu via vlm_config
|
||||
# ============================================================================
|
||||
|
||||
def _capture_generate(captured):
|
||||
def fake_post(url, json=None, timeout=None):
|
||||
captured["url"] = url
|
||||
captured["model"] = (json or {}).get("model")
|
||||
resp = MagicMock()
|
||||
resp.status_code = 200
|
||||
resp.json.return_value = {"response": '{"additional_checks": []}'}
|
||||
return resp
|
||||
return fake_post
|
||||
|
||||
|
||||
def test_safety_checks_model_default_via_vlm_config(monkeypatch):
|
||||
"""Sans RPA_SAFETY_CHECKS_LLM_MODEL, le modèle vient de vlm_config (pas gemma4 en dur)."""
|
||||
monkeypatch.delenv("RPA_SAFETY_CHECKS_LLM_MODEL", raising=False)
|
||||
captured = {}
|
||||
|
||||
import requests
|
||||
monkeypatch.setattr(requests, "post", _capture_generate(captured))
|
||||
|
||||
from agent_v0.server_v1 import safety_checks_provider as scp
|
||||
with patch.object(scp.vlm_config, "get_vlm_model", return_value="modele-resolu:test"):
|
||||
scp._call_llm_for_contextual_checks({"parameters": {}}, {}, None, [])
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
assert ":11434" in captured["url"]
|
||||
|
||||
|
||||
def test_safety_checks_model_env_override_preserved(monkeypatch):
|
||||
"""RPA_SAFETY_CHECKS_LLM_MODEL reste prioritaire (override non cassé)."""
|
||||
monkeypatch.setenv("RPA_SAFETY_CHECKS_LLM_MODEL", "mon-modele-pin")
|
||||
captured = {}
|
||||
|
||||
import requests
|
||||
monkeypatch.setattr(requests, "post", _capture_generate(captured))
|
||||
|
||||
from agent_v0.server_v1 import safety_checks_provider as scp
|
||||
# get_vlm_model ne doit pas écraser l'override : on le fait lever pour le prouver
|
||||
with patch.object(scp.vlm_config, "get_vlm_model", side_effect=AssertionError("ne doit pas être appelé")):
|
||||
scp._call_llm_for_contextual_checks({"parameters": {}}, {}, None, [])
|
||||
|
||||
assert captured["model"] == "mon-modele-pin"
|
||||
|
||||
@@ -32,6 +32,20 @@ from agent_v0.server_v1.task_planner import TaskPlanner, TaskPlan
|
||||
# Fixtures
|
||||
# =========================================================================
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _mock_vlm_model():
|
||||
"""Neutralise la résolution VLM (pas de dépendance Ollama/DGX en test).
|
||||
|
||||
Par défaut, get_vlm_model() interroge Ollama (/api/tags) ; on la fige
|
||||
pour garder les tests déterministes et hors réseau.
|
||||
"""
|
||||
with patch(
|
||||
"agent_v0.server_v1.task_planner.vlm_config.get_vlm_model",
|
||||
return_value="gemma4:latest",
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def planner():
|
||||
"""TaskPlanner avec port gemma4 factice."""
|
||||
@@ -77,6 +91,65 @@ def _mock_gemma4_response(content: str):
|
||||
return mock_resp
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# Tests : dé-hardcodage VLM (modèle via vlm_config, endpoint 11434)
|
||||
# =========================================================================
|
||||
|
||||
class TestVlmConfigDehardcode:
|
||||
"""Le modèle et l'endpoint ne doivent plus être codés en dur."""
|
||||
|
||||
def test_understand_utilise_modele_de_vlm_config(self, sample_workflows):
|
||||
"""Le payload understand() utilise le modèle résolu par vlm_config."""
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
captured["model"] = (json or {}).get("model")
|
||||
return _mock_gemma4_response("COMPRIS: NON\nWORKFLOW: AUCUN\nBOUCLE: NON\n")
|
||||
|
||||
with patch(
|
||||
"agent_v0.server_v1.task_planner.vlm_config.get_vlm_model",
|
||||
return_value="modele-resolu:test",
|
||||
), patch("requests.post", side_effect=capture_post):
|
||||
planner = TaskPlanner(domain_id="generic")
|
||||
planner.understand("Ouvre le bloc-notes", available_workflows=sample_workflows)
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
|
||||
def test_steps_to_actions_utilise_modele_de_vlm_config(self):
|
||||
"""Le payload _steps_to_actions() utilise le modèle résolu par vlm_config."""
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["model"] = (json or {}).get("model")
|
||||
return _mock_gemma4_response('{"type": "wait", "duration_ms": 100}\n')
|
||||
|
||||
with patch(
|
||||
"agent_v0.server_v1.task_planner.vlm_config.get_vlm_model",
|
||||
return_value="modele-resolu:test",
|
||||
), patch("requests.post", side_effect=capture_post):
|
||||
planner = TaskPlanner(domain_id="generic")
|
||||
planner._steps_to_actions([{"description": "1. Attendre"}], {})
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
|
||||
def test_endpoint_par_defaut_cible_11434(self, monkeypatch, sample_workflows):
|
||||
"""Sans GEMMA4_PORT, l'endpoint vise 11434 (Ollama/tunnel DGX), pas 11435."""
|
||||
monkeypatch.delenv("GEMMA4_PORT", raising=False)
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
return _mock_gemma4_response("COMPRIS: NON\nWORKFLOW: AUCUN\nBOUCLE: NON\n")
|
||||
|
||||
with patch("requests.post", side_effect=capture_post):
|
||||
planner = TaskPlanner(domain_id="generic")
|
||||
planner.understand("Ouvre le bloc-notes", available_workflows=sample_workflows)
|
||||
|
||||
assert ":11434" in captured["url"]
|
||||
assert ":11435" not in captured["url"]
|
||||
|
||||
|
||||
# =========================================================================
|
||||
# Tests : understand — ordre simple
|
||||
# =========================================================================
|
||||
|
||||
@@ -2,17 +2,55 @@
|
||||
import pytest
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
from PIL import Image
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
|
||||
from core.detection.ui_detector import UIDetector
|
||||
from core.detection.ui_detector import UIDetector, DetectionConfig
|
||||
|
||||
class TestUIDetector:
|
||||
def setup_method(self):
|
||||
self.detector = UIDetector()
|
||||
|
||||
|
||||
def test_detector_initialization(self):
|
||||
assert self.detector is not None
|
||||
|
||||
|
||||
class TestVlmModelDehardcode:
|
||||
"""Le modèle VLM ne doit plus être codé en dur (gemma4:e4b) dans la config."""
|
||||
|
||||
def test_config_no_hardcoded_gemma4(self, monkeypatch):
|
||||
"""Sans env, DetectionConfig.vlm_model ne vaut plus le littéral gemma4:e4b."""
|
||||
monkeypatch.delenv("RPA_VLM_MODEL", raising=False)
|
||||
monkeypatch.delenv("VLM_MODEL", raising=False)
|
||||
cfg = DetectionConfig()
|
||||
assert cfg.vlm_model != "gemma4:e4b"
|
||||
assert not cfg.vlm_model # None ou "" → résolution déléguée à l'init
|
||||
|
||||
def test_config_respects_env(self, monkeypatch):
|
||||
"""RPA_VLM_MODEL est honoré dans la config."""
|
||||
monkeypatch.setenv("RPA_VLM_MODEL", "mon-modele:test")
|
||||
assert DetectionConfig().vlm_model == "mon-modele:test"
|
||||
|
||||
def test_initialize_vlm_resolves_lazily(self, monkeypatch):
|
||||
"""Sans modèle explicite, _initialize_vlm résout via vlm_config (pas de hardcode)."""
|
||||
monkeypatch.delenv("RPA_VLM_MODEL", raising=False)
|
||||
monkeypatch.delenv("VLM_MODEL", raising=False)
|
||||
|
||||
captured = {}
|
||||
|
||||
class FakeClient:
|
||||
def __init__(self, endpoint=None, model=None):
|
||||
captured["model"] = model
|
||||
|
||||
import core.detection.ui_detector as uidet
|
||||
with patch.object(uidet, "check_ollama_available", return_value=True), \
|
||||
patch.object(uidet, "OllamaClient", FakeClient), \
|
||||
patch.object(uidet.vlm_config, "get_vlm_model", return_value="modele-resolu:test"):
|
||||
UIDetector()
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
pytest.main([__file__, '-v'])
|
||||
|
||||
@@ -96,6 +96,41 @@ def test_grounding_profile_gemma4_triggers_think_false(monkeypatch):
|
||||
assert p["think"] is False # gemma4 needs think=false
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────────────────
|
||||
# get_bbox_grounding_model (D5-v3b : modèle bbox_2d dédié, désambiguïsé)
|
||||
# ────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_bbox_grounding_model_default(monkeypatch):
|
||||
"""Sans env, défaut = DEFAULT_GROUNDING_FALLBACK (qwen2.5vl:7b-rpa, présent DGX)."""
|
||||
monkeypatch.delenv("RPA_BBOX_GROUNDING_MODEL", raising=False)
|
||||
monkeypatch.delenv("RPA_GROUNDING_MODEL", raising=False)
|
||||
from core.detection.vlm_config import get_bbox_grounding_model, DEFAULT_GROUNDING_FALLBACK
|
||||
|
||||
assert get_bbox_grounding_model() == DEFAULT_GROUNDING_FALLBACK == "qwen2.5vl:7b-rpa"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_bbox_grounding_model_dedicated_env(monkeypatch):
|
||||
"""RPA_BBOX_GROUNDING_MODEL est prioritaire."""
|
||||
monkeypatch.setenv("RPA_BBOX_GROUNDING_MODEL", "qwen2.5vl:32b")
|
||||
monkeypatch.setenv("RPA_GROUNDING_MODEL", "qwen3.5:9b")
|
||||
from core.detection.vlm_config import get_bbox_grounding_model
|
||||
|
||||
assert get_bbox_grounding_model() == "qwen2.5vl:32b"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_bbox_grounding_model_legacy_compat(monkeypatch):
|
||||
"""Sans la var dédiée, RPA_GROUNDING_MODEL reste honoré (rétrocompat)."""
|
||||
monkeypatch.delenv("RPA_BBOX_GROUNDING_MODEL", raising=False)
|
||||
monkeypatch.setenv("RPA_GROUNDING_MODEL", "qwen2.5vl:7b-rpa")
|
||||
from core.detection.vlm_config import get_bbox_grounding_model
|
||||
|
||||
assert get_bbox_grounding_model() == "qwen2.5vl:7b-rpa"
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────────────────
|
||||
# _extract_first_json_object
|
||||
# ────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -151,6 +151,46 @@ class TestIRBuilder:
|
||||
{"event": {"type": "key_combo", "keys": ["ctrl", "s"], "window": {"title": "*Sans titre – Bloc-notes"}, "timestamp": 115.0}},
|
||||
]
|
||||
|
||||
def test_analyze_intent_modele_via_vlm_config(self):
|
||||
"""Le payload _analyze_intent utilise le modèle résolu par vlm_config."""
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
captured["model"] = (json or {}).get("model")
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"message": {"content": "INTENTION: x\nAVANT: y\nAPRÈS: z"}}
|
||||
return resp
|
||||
|
||||
with patch(
|
||||
"core.workflow.ir_builder.vlm_config.get_vlm_model",
|
||||
return_value="modele-resolu:test",
|
||||
), patch("requests.post", side_effect=capture_post):
|
||||
builder = IRBuilder()
|
||||
builder._analyze_intent("clic Rechercher", 0, 1, "Test", "generic")
|
||||
|
||||
assert captured["model"] == "modele-resolu:test"
|
||||
|
||||
def test_analyze_intent_endpoint_par_defaut_11434(self, monkeypatch):
|
||||
"""Sans GEMMA4_PORT, _analyze_intent vise 11434, pas le port mort 11435."""
|
||||
monkeypatch.delenv("GEMMA4_PORT", raising=False)
|
||||
captured = {}
|
||||
|
||||
def capture_post(url, json=None, **kwargs):
|
||||
captured["url"] = url
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"message": {"content": "INTENTION: x\nAVANT: y\nAPRÈS: z"}}
|
||||
return resp
|
||||
|
||||
with patch("requests.post", side_effect=capture_post):
|
||||
builder = IRBuilder()
|
||||
builder._analyze_intent("clic Rechercher", 0, 1, "Test", "generic")
|
||||
|
||||
assert ":11434" in captured["url"]
|
||||
assert ":11435" not in captured["url"]
|
||||
|
||||
def test_builder_sans_gemma4(self):
|
||||
"""Le builder fonctionne même sans gemma4 (fallback gracieux)."""
|
||||
builder = IRBuilder(gemma4_port="99999") # Port invalide
|
||||
|
||||
Reference in New Issue
Block a user