File size: 5,258 Bytes
f440f03 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | """Tests for master-model routing and topology metadata."""
from unittest.mock import patch
from maris_core.orchestrator.routing import (
build_system_prompt,
detect_route,
get_specialist_branches,
)
def test_detect_route_prefers_code_branch_for_programming_requests() -> None:
route = detect_route("Lūdzu uzraksti Python API klientu ar retry loģiku")
assert route.capability == "code_generation"
assert route.branch == "coder"
assert route.profile == "coder"
assert route.target_studio == "/code"
def test_detect_route_prefers_code_branch_for_calculator_app_requests() -> None:
route = detect_route("Uzprogrammē kalkulatoru ar pogām un rezultāta lauku")
assert route.capability == "code_generation"
assert route.branch == "coder"
def test_detect_route_falls_back_to_master_chat() -> None:
route = detect_route("Pastāsti man par Latvijas ekonomiku")
assert route.capability == "text_chat"
assert route.branch == "master"
assert route.profile == "general"
def test_detect_route_keeps_programming_explanations_in_chat() -> None:
route = detect_route("Paskaidro, kā strādā REST API autentifikācija bez koda piemēra")
assert route.capability == "text_chat"
assert route.branch == "master"
def test_detect_route_does_not_send_generic_improvement_requests_to_autonomous() -> None:
route = detect_route("Uzlabo šo tekstu, lai tas būtu skaidrāks klientam")
assert route.capability == "text_chat"
assert route.branch == "master"
def test_detect_route_keeps_workflow_explanations_in_chat() -> None:
route = detect_route("Paskaidro, kā darbojas workflow orchestration Maris AI sistēmā")
assert route.capability == "text_chat"
assert route.branch == "master"
def test_detect_route_keeps_media_topic_conversations_in_chat() -> None:
route = detect_route("Es gribu vienkārši sarunāties ar Maris AI par mūziku un video idejām")
assert route.capability == "text_chat"
assert route.branch == "master"
def test_detect_route_keeps_logo_design_explanations_in_chat() -> None:
route = detect_route("Pastāsti man par logo dizaina principiem pirms ģenerēšanas")
assert route.capability == "text_chat"
assert route.branch == "master"
def test_detect_route_carries_session_context() -> None:
route = detect_route(
"Turpinām šo pašu sarunu",
session_id="session-123",
persona_id="strategist",
)
assert route.session_context.session_id == "session-123"
assert route.session_context.persona_id == "strategist"
assert route.session_context.memory_enabled is True
def test_detect_route_prefers_vision_branch_when_visual_input_exists() -> None:
route = detect_route(
"Kas ir redzams šajā attēlā?",
session_id="session-vision",
has_vision_input=True,
)
assert route.capability == "vision_analysis"
assert route.branch == "vision"
assert route.profile == "vision"
def test_detect_route_prefers_image_branch_for_logo_design_requests() -> None:
with patch.dict("os.environ", {"IMAGE_MODEL": "MarisUK/maris-ai-image"}, clear=False):
route = detect_route("Uzlabo dizainu un uztaisi jaunu logo manam produktam")
assert route.capability == "image_generation"
assert route.branch == "vision"
def test_detect_route_prefers_autonomous_branch_for_execution_style_improvement_requests() -> None:
route = detect_route("Veic labojumus un uzlabojumus manā darba plūsmā")
assert route.capability == "autonomous_tasks"
assert route.branch == "planner"
def test_build_system_prompt_changes_with_profile() -> None:
prompt = build_system_prompt("planner")
assert "plānotājs" in prompt.lower() or "prioritizē" in prompt.lower()
def test_build_system_prompt_uses_persona_specific_profile_defaults() -> None:
prompt = build_system_prompt("general", persona_id="strategist")
assert "prioritizē" in prompt.lower() or "plānotājs" in prompt.lower()
def test_specialist_branches_include_master_and_multimodal_paths() -> None:
branches = get_specialist_branches()
capabilities = {branch.capability for branch in branches}
assert "text_chat" in capabilities
assert "image_generation" in capabilities
assert "video_generation" in capabilities
def test_detect_route_uses_llm_result_when_available() -> None:
fake_pipeline = lambda messages, max_new_tokens, temperature: [ # noqa: E731
{
"generated_text": messages
+ [
{
"role": "assistant",
"content": (
'{"capability":"image_generation","reasoning":"Lietotājs prasa attēlu.",'
'"confidence":0.94}'
),
}
]
}
]
with (
patch("maris_core.text.generate.get_pipeline", return_value=fake_pipeline),
patch.dict("os.environ", {"IMAGE_MODEL": "MarisUK/maris-ai-image"}, clear=False),
):
route = detect_route("Uzzīmē man kinoplakātu ar neona gaismām")
assert route.capability == "image_generation"
assert route.branch == "vision"
assert route.profile == "image"
|