/
/
/
1"""Tests for the OpenAI Compatible plugin provider."""
2
3import json
4from types import SimpleNamespace
5from typing import TYPE_CHECKING, Any, cast
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import pytest
9from aiohttp import ClientError
10from music_assistant_models.errors import (
11 InvalidDataError,
12 LoginFailed,
13 ProviderUnavailableError,
14 RateLimited,
15 SetupFailedError,
16 UnsupportedFeaturedException,
17)
18
19from music_assistant.providers.openai_compatible import (
20 SUPPORTED_FEATURES,
21 OpenAICompatibleProvider,
22 helpers,
23 setup_flow,
24)
25from music_assistant.providers.openai_compatible.constants import (
26 CONF_MODELS,
27 SERVICE_BASE_URLS,
28 SERVICE_CUSTOM,
29)
30
31if TYPE_CHECKING:
32 from music_assistant.models.setup_flow import SetupSession
33
34BASE_URL = "https://api.example.com/v1"
35INSTANCE_ID = "openai_compatible--test_instance"
36
37# a minimal valid chat completion, for tests that only care about the request side
38_CHAT_PAYLOAD = {"choices": [{"message": {"content": "42"}}]}
39
40
41@pytest.fixture
42def mass() -> AsyncMock:
43 """Return a mocked MusicAssistant instance with a mockable HTTP session."""
44 ma = AsyncMock()
45 ma.http_session = MagicMock()
46 ma.config = MagicMock()
47 return ma
48
49
50@pytest.fixture
51def provider(mass: AsyncMock) -> OpenAICompatibleProvider:
52 """Return an OpenAICompatibleProvider with mocked dependencies and no models selected."""
53 return _build_provider(mass, {"base_url": BASE_URL})
54
55
56async def test_list_models_returns_sorted_ids(mass: AsyncMock) -> None:
57 """The discovered model ids come back sorted."""
58 mass.http_session.request = MagicMock(
59 return_value=_response_cm(
60 response=_json_response(200, {"data": [{"id": "c"}, {"id": "a"}, {"id": "b"}]})
61 )
62 )
63
64 assert await helpers.list_models(mass, BASE_URL, "", timeout=10) == ["a", "b", "c"]
65
66
67async def test_list_models_skips_entries_without_a_usable_id(mass: AsyncMock) -> None:
68 """Entries missing an id, or with a blank/non-string one, are dropped rather than raising."""
69 mass.http_session.request = MagicMock(
70 return_value=_response_cm(
71 response=_json_response(
72 200,
73 {
74 "data": [
75 {"id": "good-model"},
76 {"id": ""},
77 {"id": 123},
78 {"no_id": "x"},
79 "not-a-dict",
80 ]
81 },
82 )
83 )
84 )
85
86 assert await helpers.list_models(mass, BASE_URL, "", timeout=10) == ["good-model"]
87
88
89async def test_list_models_raises_on_an_http_error(mass: AsyncMock) -> None:
90 """A 404 usually means a wrong address, so it must not pass as "nothing to offer"."""
91 mass.http_session.request = MagicMock(
92 return_value=_response_cm(response=_json_response(404, {}))
93 )
94
95 with pytest.raises(InvalidDataError):
96 await helpers.list_models(mass, BASE_URL, "", timeout=10)
97
98
99async def test_list_models_returns_empty_list_when_the_answer_carries_no_list(
100 mass: AsyncMock,
101) -> None:
102 """An endpoint answering at the right address without a list has nothing to offer."""
103 mass.http_session.request = MagicMock(
104 return_value=_response_cm(response=_json_response(200, {}))
105 )
106
107 assert await helpers.list_models(mass, BASE_URL, "", timeout=10) == []
108
109
110async def test_list_models_raises_login_failed_on_401(mass: AsyncMock) -> None:
111 """Bad credentials must be reported, not swallowed into an empty list."""
112 mass.http_session.request = MagicMock(
113 return_value=_response_cm(response=_json_response(401, {}))
114 )
115
116 with pytest.raises(LoginFailed):
117 await helpers.list_models(mass, BASE_URL, "bad-key", timeout=10)
118
119
120async def test_login_failed_message_includes_the_upstream_error_detail(mass: AsyncMock) -> None:
121 """The service's own error message reaches the user, not just its status code."""
122 body = {"error": {"message": "Incorrect API key provided"}}
123 mass.http_session.request = MagicMock(
124 return_value=_response_cm(response=_json_response(401, body))
125 )
126
127 with pytest.raises(LoginFailed, match="Incorrect API key provided"):
128 await helpers.list_models(mass, BASE_URL, "bad-key", timeout=10)
129
130
131async def test_list_models_raises_provider_unavailable_on_connection_error(
132 mass: AsyncMock,
133) -> None:
134 """An unreachable host must be reported, not swallowed into an empty list."""
135 mass.http_session.request = MagicMock(
136 return_value=_response_cm(exc=ClientError("connection refused"))
137 )
138
139 with pytest.raises(ProviderUnavailableError):
140 await helpers.list_models(mass, BASE_URL, "", timeout=10)
141
142
143async def test_list_models_raises_provider_unavailable_on_timeout(mass: AsyncMock) -> None:
144 """A hung request must be reported, not swallowed into an empty list."""
145 mass.http_session.request = MagicMock(return_value=_response_cm(exc=TimeoutError()))
146
147 with pytest.raises(ProviderUnavailableError):
148 await helpers.list_models(mass, BASE_URL, "", timeout=10)
149
150
151async def test_chat_completion_returns_trimmed_content(mass: AsyncMock) -> None:
152 """The answer is returned with surrounding whitespace trimmed."""
153 mass.http_session.request = MagicMock(
154 return_value=_response_cm(
155 response=_json_response(
156 200, {"choices": [{"message": {"content": " The answer is 42. "}}]}
157 )
158 )
159 )
160
161 result = await helpers.chat_completion(
162 mass, BASE_URL, "", model="my-model", prompt="What is the answer?", timeout=30
163 )
164
165 assert result == "The answer is 42."
166
167
168async def test_chat_completion_strips_leading_think_block(mass: AsyncMock) -> None:
169 """A reasoning model's <think> preamble is removed, leaving only the answer."""
170 content = "<think>Let me reason about this.</think>Paris is the capital of France."
171 mass.http_session.request = MagicMock(
172 return_value=_response_cm(
173 response=_json_response(200, {"choices": [{"message": {"content": content}}]})
174 )
175 )
176
177 result = await helpers.chat_completion(
178 mass, BASE_URL, "", model="my-model", prompt="capital of France?", timeout=30
179 )
180
181 assert result == "Paris is the capital of France."
182
183
184async def test_chat_completion_raises_on_think_only_content(mass: AsyncMock) -> None:
185 """A reply that is only a think block leaves no answer, which is an error."""
186 content = "<think>Still thinking about it...</think>"
187 mass.http_session.request = MagicMock(
188 return_value=_response_cm(
189 response=_json_response(200, {"choices": [{"message": {"content": content}}]})
190 )
191 )
192
193 with pytest.raises(InvalidDataError):
194 await helpers.chat_completion(mass, BASE_URL, "", model="my-model", prompt="hi", timeout=30)
195
196
197async def test_chat_completion_raises_on_blank_content(mass: AsyncMock) -> None:
198 """Whitespace-only content is treated the same as no answer at all."""
199 mass.http_session.request = MagicMock(
200 return_value=_response_cm(
201 response=_json_response(200, {"choices": [{"message": {"content": " "}}]})
202 )
203 )
204
205 with pytest.raises(InvalidDataError):
206 await helpers.chat_completion(mass, BASE_URL, "", model="my-model", prompt="hi", timeout=30)
207
208
209async def test_chat_completion_raises_when_choices_missing(mass: AsyncMock) -> None:
210 """A response without a choices list is malformed, not a valid empty answer."""
211 mass.http_session.request = MagicMock(
212 return_value=_response_cm(response=_json_response(200, {}))
213 )
214
215 with pytest.raises(InvalidDataError):
216 await helpers.chat_completion(mass, BASE_URL, "", model="my-model", prompt="hi", timeout=30)
217
218
219async def test_chat_completion_raises_rate_limited_with_retry_after(mass: AsyncMock) -> None:
220 """A 429 is raised as RateLimited, honouring a numeric Retry-After as the backoff time."""
221 mass.http_session.request = MagicMock(
222 return_value=_response_cm(response=_json_response(429, {}, retry_after="45"))
223 )
224
225 with pytest.raises(RateLimited) as excinfo:
226 await helpers.chat_completion(mass, BASE_URL, "", model="my-model", prompt="hi", timeout=30)
227
228 assert excinfo.value.backoff_time == 45
229
230
231async def test_chat_completion_request_body_has_no_max_tokens(mass: AsyncMock) -> None:
232 """The model and prompt are sent as a single user message, deliberately without max_tokens."""
233 mass.http_session.request = MagicMock(
234 return_value=_response_cm(response=_json_response(200, _CHAT_PAYLOAD))
235 )
236
237 await helpers.chat_completion(
238 mass, BASE_URL, "", model="my-model", prompt="What is up?", timeout=30
239 )
240
241 sent = mass.http_session.request.call_args.kwargs["json"]
242 assert sent["model"] == "my-model"
243 assert sent["messages"] == [{"role": "user", "content": "What is up?"}]
244 assert "max_tokens" not in sent
245
246
247def test_error_detail_falls_back_to_raw_text_for_non_json_body() -> None:
248 """A body that is not JSON at all still yields something readable."""
249 assert helpers._error_detail(b"Service Unavailable") == "Service Unavailable"
250
251
252def test_error_detail_returns_placeholder_for_empty_body() -> None:
253 """An empty response body must not surface as an empty, unhelpful error message."""
254 assert helpers._error_detail(b"") == "no details provided"
255
256
257def test_error_detail_truncates_overlong_messages() -> None:
258 """An unbounded upstream message is capped so it cannot bloat the raised error."""
259 long_message = "x" * (helpers.MAX_ERROR_DETAIL_CHARS + 50)
260 body = json.dumps({"error": {"message": long_message}}).encode()
261
262 result = helpers._error_detail(body)
263
264 assert result == long_message[: helpers.MAX_ERROR_DETAIL_CHARS]
265
266
267def test_base_url_strips_trailing_slash() -> None:
268 """Every request path is built by appending to the stored base URL."""
269 provider = _provider_for_setup({"base_url": "https://api.example.com/v1/"})
270
271 assert provider._base_url == "https://api.example.com/v1"
272
273
274def test_base_url_strips_whitespace() -> None:
275 """Whitespace around a stored base URL is stripped."""
276 provider = _provider_for_setup({"base_url": " https://api.example.com/v1 "})
277
278 assert provider._base_url == "https://api.example.com/v1"
279
280
281async def test_handle_async_init_accepts_a_configured_base_url() -> None:
282 """A usable endpoint must pass the init check, whatever shape the address has."""
283 provider = _provider_for_setup({"base_url": "http://localhost:11434/v1"})
284
285 await provider.handle_async_init()
286
287
288async def test_handle_async_init_raises_when_base_url_missing() -> None:
289 """A never-configured base URL cannot be initialized into a working provider."""
290 provider = _provider_for_setup({})
291
292 with pytest.raises(SetupFailedError):
293 await provider.handle_async_init()
294
295
296async def test_handle_async_init_raises_when_base_url_blank() -> None:
297 """A blank base URL is treated the same as a missing one."""
298 provider = _provider_for_setup({"base_url": " "})
299
300 with pytest.raises(SetupFailedError):
301 await provider.handle_async_init()
302
303
304def test_api_key_defaults_to_empty_string_when_not_stored() -> None:
305 """No stored key means the no-auth local-server case, not None or the text "None"."""
306 provider = _provider_for_setup({"base_url": BASE_URL})
307
308 assert provider._api_key == ""
309
310
311def test_api_key_comes_from_setup_data_and_is_stripped() -> None:
312 """The key is only stored in setup_data, and surrounding whitespace is removed."""
313 provider = _provider_for_setup({"base_url": BASE_URL, "api_key": " sk-secret "})
314
315 assert provider._api_key == "sk-secret"
316
317
318async def test_probe_returns_invalid_api_key_slug_on_login_failed() -> None:
319 """A rejected key must map to the frontend's invalid_api_key error slug."""
320 session = cast("SetupSession", SimpleNamespace(mass=MagicMock()))
321 with patch.object(setup_flow, "list_models", AsyncMock(side_effect=LoginFailed("bad key"))):
322 result = await setup_flow._probe(session, {"base_url": BASE_URL, "api_key": "bad"})
323
324 assert result == "invalid_api_key"
325
326
327async def test_probe_returns_cannot_connect_slug_on_provider_unavailable() -> None:
328 """An unreachable host must map to the frontend's cannot_connect error slug."""
329 session = cast("SetupSession", SimpleNamespace(mass=MagicMock()))
330 with patch.object(
331 setup_flow, "list_models", AsyncMock(side_effect=ProviderUnavailableError("down"))
332 ):
333 result = await setup_flow._probe(session, {"base_url": BASE_URL})
334
335 assert result == "cannot_connect"
336
337
338async def test_probe_succeeds_when_endpoint_has_no_model_listing() -> None:
339 """An empty model listing is a valid, working endpoint, not a probe failure."""
340 session = cast("SetupSession", SimpleNamespace(mass=MagicMock()))
341 with patch.object(setup_flow, "list_models", AsyncMock(return_value=[])):
342 result = await setup_flow._probe(session, {"base_url": BASE_URL})
343
344 assert result is None
345
346
347async def test_probe_reports_an_address_that_answers_with_an_error() -> None:
348 """A wrong address must be caught during setup, not on the first AI query."""
349 session = cast("SetupSession", SimpleNamespace(mass=MagicMock()))
350 with patch.object(setup_flow, "list_models", AsyncMock(side_effect=InvalidDataError("404"))):
351 result = await setup_flow._probe(session, {"base_url": BASE_URL})
352
353 assert result == "cannot_connect"
354
355
356async def test_run_setup_prefills_the_address_of_the_chosen_service() -> None:
357 """Picking a known service fills in its address so it need not be looked up."""
358 session = _FakeSession(setup_data={}, submissions=[{"service": "groq"}, {}])
359
360 await _run_setup(session)
361
362 assert _entry_value(session, "base_url") == SERVICE_BASE_URLS["groq"]
363
364
365async def test_run_setup_leaves_the_address_empty_for_a_custom_service() -> None:
366 """There is nothing to prefill for a service we do not know the address of."""
367 session = _FakeSession(setup_data={}, submissions=[{"service": SERVICE_CUSTOM}, {}])
368
369 await _run_setup(session)
370
371 assert _entry_value(session, "base_url") == ""
372
373
374async def test_run_setup_prefers_the_new_services_address_when_switching() -> None:
375 """Moving an instance to another service must not keep the previous address."""
376 session = _FakeSession(
377 setup_data={"service": "openai", "base_url": SERVICE_BASE_URLS["openai"]},
378 submissions=[{"service": "ollama"}, {}],
379 )
380
381 await _run_setup(session)
382
383 assert _entry_value(session, "base_url") == SERVICE_BASE_URLS["ollama"]
384
385
386async def test_run_setup_keeps_an_edited_address_for_the_same_service() -> None:
387 """A hand-corrected address survives a reconfigure that keeps the same service."""
388 edited = "http://nas.local:11434/v1"
389 session = _FakeSession(
390 setup_data={"service": "ollama", "base_url": edited},
391 submissions=[{"service": "ollama"}, {}],
392 )
393
394 await _run_setup(session)
395
396 assert _entry_value(session, "base_url") == edited
397
398
399async def test_run_setup_keeps_the_stored_api_key_when_the_field_is_left_empty() -> None:
400 """The stored key is never sent to the client, so an empty field must not wipe it."""
401 session = _FakeSession(
402 setup_data={"service": "openai", "base_url": BASE_URL, "api_key": "sk-existing"},
403 submissions=[{"service": "openai"}, {"base_url": BASE_URL, "api_key": ""}],
404 )
405
406 await _run_setup(session)
407
408 assert session.finished == {
409 "service": "openai",
410 "base_url": BASE_URL,
411 "api_key": "sk-existing",
412 }
413
414
415async def test_run_setup_replaces_the_stored_api_key_when_a_new_one_is_given() -> None:
416 """A freshly typed key overwrites the stored one."""
417 session = _FakeSession(
418 setup_data={"service": "openai", "base_url": BASE_URL, "api_key": "sk-existing"},
419 submissions=[{"service": "openai"}, {"base_url": BASE_URL, "api_key": "sk-new"}],
420 )
421
422 await _run_setup(session)
423
424 assert session.finished == {"service": "openai", "base_url": BASE_URL, "api_key": "sk-new"}
425
426
427async def test_run_setup_clears_the_stored_api_key_on_request() -> None:
428 """Ticking the clear option removes the key, for a move to a keyless server."""
429 session = _FakeSession(
430 setup_data={"service": "openai", "base_url": BASE_URL, "api_key": "sk-existing"},
431 submissions=[
432 {"service": "ollama"},
433 {"base_url": BASE_URL, "api_key": "", "clear_api_key": True},
434 ],
435 )
436
437 await _run_setup(session)
438
439 assert session.finished == {"service": "ollama", "base_url": BASE_URL, "api_key": ""}
440
441
442async def test_run_setup_offers_the_clear_option_only_when_a_key_is_stored() -> None:
443 """A first-time setup has nothing to clear, so the option stays out of the form."""
444 fresh = _FakeSession(setup_data={}, submissions=[{"service": "openai"}, {}])
445 stored = _FakeSession(
446 setup_data={"service": "openai", "base_url": BASE_URL, "api_key": "sk-existing"},
447 submissions=[{"service": "openai"}, {"base_url": BASE_URL, "api_key": ""}],
448 )
449
450 await _run_setup(fresh)
451 await _run_setup(stored)
452
453 assert setup_flow.CONF_CLEAR_API_KEY not in [entry.key for entry in fresh.entries]
454 assert setup_flow.CONF_CLEAR_API_KEY in [entry.key for entry in stored.entries]
455
456
457async def test_get_ai_engines_returns_one_engine_per_configured_model(
458 provider: OpenAICompatibleProvider,
459) -> None:
460 """Each configured model becomes its own engine, keyed by its bare model id."""
461 _configure_models(provider, ["model-a", "model-b"])
462
463 engines = await provider.get_ai_engines()
464
465 assert [engine.id for engine in engines] == ["model-a", "model-b"]
466 assert all(engine.id == engine.name for engine in engines)
467 assert [engine.uid for engine in engines] == [
468 f"{INSTANCE_ID}/model-a",
469 f"{INSTANCE_ID}/model-b",
470 ]
471
472
473async def test_get_ai_engines_empty_when_no_models_configured(
474 provider: OpenAICompatibleProvider,
475) -> None:
476 """No selection yields no engines rather than a default model."""
477 _configure_models(provider, [])
478
479 assert await provider.get_ai_engines() == []
480
481
482async def test_get_ai_engines_empty_when_config_value_absent(
483 provider: OpenAICompatibleProvider,
484) -> None:
485 """A provider whose options form was never saved yields no engines instead of raising."""
486 provider.config.get_value = MagicMock( # type: ignore[method-assign]
487 side_effect=lambda _key, default=None: default
488 )
489
490 assert await provider.get_ai_engines() == []
491
492
493async def test_get_ai_engines_ignores_non_string_junk(
494 provider: OpenAICompatibleProvider,
495) -> None:
496 """Malformed stored entries are dropped instead of raising or producing bad engines."""
497 _configure_models(provider, ["model-a", 123, None, " ", ""])
498
499 engines = await provider.get_ai_engines()
500
501 assert [engine.id for engine in engines] == ["model-a"]
502
503
504async def test_ai_query_uses_the_given_engine_id(
505 provider: OpenAICompatibleProvider, mass: AsyncMock
506) -> None:
507 """An explicit engine id is the model sent in the request, regardless of the selection."""
508 _configure_models(provider, ["model-a", "model-b"])
509 mass.http_session.request = MagicMock(
510 return_value=_response_cm(response=_json_response(200, _CHAT_PAYLOAD))
511 )
512
513 await provider.ai_query("hello", engine_id="model-b")
514
515 assert mass.http_session.request.call_args.kwargs["json"]["model"] == "model-b"
516
517
518async def test_ai_query_falls_back_to_first_configured_model(
519 provider: OpenAICompatibleProvider, mass: AsyncMock
520) -> None:
521 """No engine id falls back to the first configured model, in sorted order."""
522 _configure_models(provider, ["model-b", "model-a"])
523 mass.http_session.request = MagicMock(
524 return_value=_response_cm(response=_json_response(200, _CHAT_PAYLOAD))
525 )
526
527 await provider.ai_query("hello", engine_id=None)
528
529 assert mass.http_session.request.call_args.kwargs["json"]["model"] == "model-a"
530
531
532async def test_ai_query_raises_when_no_models_configured(
533 provider: OpenAICompatibleProvider,
534) -> None:
535 """No selected model at all is a hard failure, not a silent default."""
536 _configure_models(provider, [])
537
538 with pytest.raises(UnsupportedFeaturedException):
539 await provider.ai_query("hello")
540
541
542async def test_get_config_entries_models_option_union_survives_vanished_selection(
543 provider: OpenAICompatibleProvider, mass: AsyncMock
544) -> None:
545 """A selected model missing from the endpoint's current listing is still offered."""
546 _configure_models(provider, ["stale-model"])
547 mass.http_session.request = MagicMock(
548 return_value=_response_cm(response=_json_response(200, {"data": [{"id": "fresh-model"}]}))
549 )
550
551 entries = await provider.get_config_entries()
552
553 models_entry = next(entry for entry in entries if entry.key == CONF_MODELS)
554 assert models_entry.multi_value is True
555 assert {option.value for option in models_entry.options} == {"stale-model", "fresh-model"}
556
557
558async def test_get_config_entries_options_empty_when_discovery_fails(
559 provider: OpenAICompatibleProvider, mass: AsyncMock
560) -> None:
561 """A failed listing leaves no options, which is what lets the frontend accept free text."""
562 _configure_models(provider, [])
563 mass.http_session.request = MagicMock(
564 return_value=_response_cm(exc=ClientError("connection refused"))
565 )
566
567 entries = await provider.get_config_entries()
568
569 models_entry = next(entry for entry in entries if entry.key == CONF_MODELS)
570 assert models_entry.options == []
571
572
573async def test_get_config_entries_survives_an_endpoint_without_a_listing(
574 provider: OpenAICompatibleProvider, mass: AsyncMock
575) -> None:
576 """Setup rejects a listing error, but an already-configured provider must still load."""
577 _configure_models(provider, ["typed-by-hand"])
578 mass.http_session.request = MagicMock(
579 return_value=_response_cm(response=_json_response(404, {}))
580 )
581
582 entries = await provider.get_config_entries()
583
584 models_entry = next(entry for entry in entries if entry.key == CONF_MODELS)
585 assert [option.value for option in models_entry.options] == ["typed-by-hand"]
586
587
588async def test_get_config_entries_reaches_the_endpoint_without_async_init(
589 provider: OpenAICompatibleProvider, mass: AsyncMock
590) -> None:
591 """Loading resolves the config entries before async init, so they must not depend on it."""
592 _configure_models(provider, [])
593 mass.http_session.request = MagicMock(
594 return_value=_response_cm(response=_json_response(200, {"data": [{"id": "fresh-model"}]}))
595 )
596
597 # deliberately no handle_async_init(): mass._load_provider only runs it afterwards
598 entries = await provider.get_config_entries()
599
600 assert mass.http_session.request.call_args.args[1] == f"{BASE_URL}/models"
601 models_entry = next(entry for entry in entries if entry.key == CONF_MODELS)
602 assert [option.value for option in models_entry.options] == ["fresh-model"]
603
604
605def _response_cm(*, response: MagicMock | None = None, exc: Exception | None = None) -> MagicMock:
606 """Build a fake async context manager mimicking aiohttp's session.request()."""
607 cm = MagicMock()
608 cm.__aenter__ = AsyncMock(return_value=response, side_effect=exc)
609 cm.__aexit__ = AsyncMock(return_value=False)
610 return cm
611
612
613def _json_response(status: int, payload: dict[str, Any], *, retry_after: str = "") -> MagicMock:
614 """Build a fake aiohttp response exposing the attributes _request reads."""
615 response = MagicMock()
616 response.status = status
617 response.headers = {"Retry-After": retry_after} if retry_after else {}
618 response.read = AsyncMock(return_value=json.dumps(payload).encode())
619 return response
620
621
622def _configure_models(provider: OpenAICompatibleProvider, models: list[Any]) -> None:
623 """Make the provider report the given stored value for the models config entry."""
624 provider.config.get_value = MagicMock( # type: ignore[method-assign]
625 side_effect=lambda key, default=None: models if key == CONF_MODELS else default
626 )
627
628
629def _provider_for_setup(setup_data: dict[str, Any]) -> OpenAICompatibleProvider:
630 """Build a provider backed by the given setup_data, with nothing stored in its options."""
631 mass = AsyncMock()
632 mass.http_session = MagicMock()
633 mass.config = MagicMock()
634 return _build_provider(mass, setup_data)
635
636
637def _build_provider(mass: AsyncMock, setup_data: dict[str, Any]) -> OpenAICompatibleProvider:
638 """Build a provider on the given mass, backed by the given setup_data."""
639 mass.config.get = MagicMock(return_value=dict(setup_data))
640 mass.config.decrypt_string = MagicMock(side_effect=lambda value: value)
641 manifest = MagicMock()
642 manifest.domain = "openai_compatible"
643 config = MagicMock()
644 config.values = {}
645 config.instance_id = INSTANCE_ID
646 config.get_value = MagicMock(return_value="GLOBAL")
647 provider = OpenAICompatibleProvider(mass, manifest, config, SUPPORTED_FEATURES)
648 # the log level is read during construction; past that nothing is stored in the options
649 config.get_value = MagicMock(side_effect=lambda _key, default=None: default)
650 return provider
651
652
653def _entry_value(session: _FakeSession, key: str) -> Any:
654 """Return the prefilled value of an entry on the last form the flow rendered."""
655 return next(entry.value for entry in session.entries if entry.key == key)
656
657
658async def _run_setup(session: _FakeSession) -> None:
659 """Drive the setup flow against a fake session, with a reachable endpoint."""
660 with patch.object(setup_flow, "list_models", AsyncMock(return_value=[])):
661 await setup_flow.run_setup(cast("SetupSession", session))
662
663
664class _FakeSession:
665 """Minimal stand-in for SetupSession that replays scripted form submissions."""
666
667 def __init__(self, setup_data: dict[str, Any], submissions: list[dict[str, Any]]) -> None:
668 """Initialize with the stored setup_data and the submissions to replay in order."""
669 self.context = SimpleNamespace(setup_data=setup_data)
670 self.mass = MagicMock()
671 self.entries: list[Any] = []
672 self.finished: dict[str, Any] | None = None
673 self._submissions = list(submissions)
674
675 async def form(self, entries: list[Any], **kwargs: Any) -> dict[str, Any]:
676 """Record the rendered entries and return the next scripted submission."""
677 self.entries = entries
678 return dict(self._submissions.pop(0))
679
680 async def finish(self, values: dict[str, Any]) -> dict[str, str]:
681 """Record the values the flow decided to persist."""
682 self.finished = dict(values)
683 return {"instance_id": "openai_compatible--new"}
684