/
/
/
1"""Tests for the DirectConnectionHandler (direct connection mode)."""
2
3from __future__ import annotations
4
5import json
6import time
7from typing import TYPE_CHECKING, Any
8from unittest.mock import AsyncMock, MagicMock, patch
9
10import pytest
11from aiohttp.test_utils import make_mocked_request
12
13if TYPE_CHECKING:
14 from aiohttp import web
15
16from music_assistant.providers.yandex_smarthome.constants import (
17 CONNECTION_TYPE_DIRECT,
18 DIRECT_API_BASE_PATH,
19 DIRECT_AUTH_BASE_PATH,
20 DIRECT_HEALTH_RESPONSE,
21 DIRECT_OAUTH_CLIENT_ID,
22 OAUTH_CODE_EXPIRY,
23)
24from music_assistant.providers.yandex_smarthome.direct import DirectConnectionHandler
25from music_assistant.providers.yandex_smarthome.plugin import YandexSmartHomePlugin
26
27
28def _body_json(resp: web.Response) -> Any:
29 """Decode the JSON body of a response."""
30 assert isinstance(resp.body, bytes | bytearray)
31 return json.loads(resp.body)
32
33
34TEST_CLIENT_SECRET = "test-client-secret-abc123"
35
36# token_store lists shared between fixtures and tests
37_handler_tokens: list[str] = []
38_handler_no_token_tokens: list[str] = []
39
40
41# ---------------------------------------------------------------------------
42# Fixtures
43# ---------------------------------------------------------------------------
44
45
46@pytest.fixture
47def mock_mass() -> MagicMock:
48 """Return a mock MusicAssistant with a webserver stub."""
49 mass = MagicMock()
50 mass.webserver.base_url = "https://my-ma.example.com"
51 mass.webserver.register_dynamic_route = MagicMock(return_value=MagicMock())
52 mass.players = []
53 mass.http_session = MagicMock()
54 # No setup_data persisted on the mock config store, so get_setup_value falls through
55 # to the config entry value/default via get_config_value (the provided ProviderConfig).
56 mass.config.get = MagicMock(return_value=None)
57 mass.config.get_raw_provider_config_value = MagicMock(return_value=None)
58 return mass
59
60
61@pytest.fixture
62def handler(mock_mass: MagicMock) -> DirectConnectionHandler:
63 """Return a DirectConnectionHandler with a known token."""
64 _handler_tokens.clear()
65
66 def on_token(t: str) -> None:
67 _handler_tokens.append(t)
68
69 return DirectConnectionHandler(
70 mass=mock_mass,
71 user_id="test_user",
72 access_token="test-token-abc",
73 client_secret=TEST_CLIENT_SECRET,
74 exposed_ids=None,
75 on_token_created=on_token,
76 )
77
78
79@pytest.fixture
80def handler_no_token(mock_mass: MagicMock) -> DirectConnectionHandler:
81 """Return a handler with no initial access token (first-time OAuth flow)."""
82 _handler_no_token_tokens.clear()
83
84 def on_token(t: str) -> None:
85 _handler_no_token_tokens.append(t)
86
87 return DirectConnectionHandler(
88 mass=mock_mass,
89 user_id="test_user",
90 access_token="",
91 client_secret=TEST_CLIENT_SECRET,
92 exposed_ids=None,
93 on_token_created=on_token,
94 )
95
96
97def _make_request(
98 method: str = "GET",
99 path: str = "/",
100 headers: dict[str, str] | None = None,
101 payload: dict[str, Any] | None = None,
102 query: dict[str, str] | None = None,
103 post_data: dict[str, str] | None = None,
104) -> web.Request:
105 """Build a mock aiohttp Request."""
106 req = make_mocked_request(
107 method,
108 path,
109 headers=headers or {},
110 )
111 if query:
112 object.__setattr__(req, "_rel_url", req._rel_url.with_query(query))
113 if payload is not None:
114 object.__setattr__(req, "_payload_writer", None)
115
116 async def _json(**_kwargs: Any) -> dict[str, Any]:
117 return payload
118
119 object.__setattr__(req, "json", _json)
120 if post_data is not None:
121
122 async def _post() -> dict[str, str]:
123 return post_data
124
125 object.__setattr__(req, "post", _post)
126 return req
127
128
129def _get_pending_code(h: DirectConnectionHandler) -> str:
130 """Return the first pending authorization code from a handler."""
131 return str(next(iter(h._pending_codes.keys())))
132
133
134# ---------------------------------------------------------------------------
135# Constants
136# ---------------------------------------------------------------------------
137
138
139def test_connection_type_direct() -> None:
140 """CONNECTION_TYPE_DIRECT should be 'direct'."""
141 assert CONNECTION_TYPE_DIRECT == "direct"
142
143
144def test_api_base_path() -> None:
145 """DIRECT_API_BASE_PATH should be under /api/yandex_smarthome."""
146 assert DIRECT_API_BASE_PATH.startswith("/api/yandex_smarthome")
147
148
149def test_auth_base_path() -> None:
150 """DIRECT_AUTH_BASE_PATH should be under /api/yandex_smarthome."""
151 assert DIRECT_AUTH_BASE_PATH.startswith("/api/yandex_smarthome")
152
153
154def test_health_response() -> None:
155 """DIRECT_HEALTH_RESPONSE should be non-empty."""
156 assert len(DIRECT_HEALTH_RESPONSE) > 0
157
158
159def test_oauth_constants() -> None:
160 """OAuth constants should match Yandex Smart Home spec."""
161 assert DIRECT_OAUTH_CLIENT_ID == "https://social.yandex.net/"
162 assert OAUTH_CODE_EXPIRY == 300
163
164
165# ---------------------------------------------------------------------------
166# Route registration
167# ---------------------------------------------------------------------------
168
169
170def test_register_routes(handler: DirectConnectionHandler, mock_mass: MagicMock) -> None:
171 """register_routes should register all 10 HTTP routes."""
172 handler.register_routes()
173 assert mock_mass.webserver.register_dynamic_route.call_count == 10
174
175
176def test_unregister_routes(handler: DirectConnectionHandler) -> None:
177 """unregister_routes should call all stored callbacks and clear."""
178 handler.register_routes()
179 handler.unregister_routes()
180 assert len(handler._unregister_callbacks) == 0
181
182
183def test_register_routes_rolls_back_on_failure(
184 handler: DirectConnectionHandler, mock_mass: MagicMock
185) -> None:
186 """register_routes must unregister partial routes and re-raise on RuntimeError."""
187 unregister_cb = MagicMock()
188 call_count = {"n": 0}
189
190 def register(_path: str, _handler_fn: Any, _method: str) -> Any:
191 call_count["n"] += 1
192 if call_count["n"] == 3:
193 raise RuntimeError("already registered")
194 return unregister_cb
195
196 mock_mass.webserver.register_dynamic_route.side_effect = register
197
198 with pytest.raises(RuntimeError):
199 handler.register_routes()
200
201 # 2 successful registrations must be rolled back via unregister_cb
202 assert unregister_cb.call_count == 2
203 assert handler._unregister_callbacks == []
204
205
206# ---------------------------------------------------------------------------
207# Auth validation
208# ---------------------------------------------------------------------------
209
210
211def test_auth_valid(handler: DirectConnectionHandler) -> None:
212 """Valid Bearer token should pass validation."""
213 req = _make_request(headers={"Authorization": "Bearer test-token-abc"})
214 assert handler._validate_auth(req) is True
215
216
217def test_auth_invalid(handler: DirectConnectionHandler) -> None:
218 """Wrong Bearer token should fail validation."""
219 req = _make_request(headers={"Authorization": "Bearer wrong-token"})
220 assert handler._validate_auth(req) is False
221
222
223def test_auth_missing(handler: DirectConnectionHandler) -> None:
224 """Missing Authorization header should fail validation."""
225 req = _make_request()
226 assert handler._validate_auth(req) is False
227
228
229def test_auth_non_bearer(handler: DirectConnectionHandler) -> None:
230 """Non-Bearer auth scheme should fail validation."""
231 req = _make_request(headers={"Authorization": "Basic dXNlcjpwYXNz"})
232 assert handler._validate_auth(req) is False
233
234
235def test_auth_empty_token_rejects(handler_no_token: DirectConnectionHandler) -> None:
236 """Handler with no access token should reject any Bearer token."""
237 req = _make_request(headers={"Authorization": "Bearer anything"})
238 assert handler_no_token._validate_auth(req) is False
239
240
241# ---------------------------------------------------------------------------
242# Health check
243# ---------------------------------------------------------------------------
244
245
246@pytest.mark.asyncio
247async def test_health_get(handler: DirectConnectionHandler) -> None:
248 """GET health check should return 200 with health text."""
249 req = _make_request(method="GET", path="/v1.0")
250 resp = await handler._handle_health(req)
251 assert resp.status == 200
252 assert resp.text == DIRECT_HEALTH_RESPONSE
253
254
255@pytest.mark.asyncio
256async def test_health_head(handler: DirectConnectionHandler) -> None:
257 """HEAD health check should return 200."""
258 req = _make_request(method="HEAD", path="/v1.0")
259 resp = await handler._handle_health(req)
260 assert resp.status == 200
261
262
263# ---------------------------------------------------------------------------
264# API auth rejection
265# ---------------------------------------------------------------------------
266
267
268@pytest.mark.asyncio
269async def test_devices_unauthorized(handler: DirectConnectionHandler) -> None:
270 """POST /user/devices with bad token should return 401."""
271 req = _make_request(method="POST", headers={"Authorization": "Bearer bad"})
272 resp = await handler._handle_devices(req)
273 assert resp.status == 401
274
275
276@pytest.mark.asyncio
277async def test_query_unauthorized(handler: DirectConnectionHandler) -> None:
278 """POST /user/devices/query with bad token should return 401."""
279 req = _make_request(method="POST", headers={"Authorization": "Bearer bad"})
280 resp = await handler._handle_query(req)
281 assert resp.status == 401
282
283
284@pytest.mark.asyncio
285async def test_action_unauthorized(handler: DirectConnectionHandler) -> None:
286 """POST /user/devices/action with bad token should return 401."""
287 req = _make_request(method="POST", headers={"Authorization": "Bearer bad"})
288 resp = await handler._handle_action(req)
289 assert resp.status == 401
290
291
292@pytest.mark.asyncio
293async def test_unlink_unauthorized(handler: DirectConnectionHandler) -> None:
294 """POST /user/unlink with bad token should return 401."""
295 req = _make_request(method="POST", headers={"Authorization": "Bearer bad"})
296 resp = await handler._handle_unlink(req)
297 assert resp.status == 401
298
299
300# ---------------------------------------------------------------------------
301# API authorized calls
302# ---------------------------------------------------------------------------
303
304_AUTH_HEADERS = {"Authorization": "Bearer test-token-abc"}
305
306
307@pytest.mark.asyncio
308async def test_devices_success(handler: DirectConnectionHandler) -> None:
309 """Authorized /user/devices should return 200 with device list."""
310 req = _make_request(
311 method="POST",
312 headers={**_AUTH_HEADERS, "X-Request-Id": "req-1"},
313 )
314 mock_result = MagicMock()
315 resp_payload = {"request_id": "req-1", "payload": {"devices": []}}
316 mock_hdl = patch(
317 "music_assistant.providers.yandex_smarthome.direct.handle_device_list",
318 new_callable=AsyncMock,
319 return_value=mock_result,
320 )
321 with (
322 mock_hdl,
323 patch(
324 "music_assistant.providers.yandex_smarthome.direct.asdict", return_value={"devices": []}
325 ),
326 patch(
327 "music_assistant.providers.yandex_smarthome.direct.build_response",
328 return_value=resp_payload,
329 ),
330 ):
331 resp = await handler._handle_devices(req)
332 assert resp.status == 200
333 body = _body_json(resp)
334 assert body["request_id"] == "req-1"
335
336
337@pytest.mark.asyncio
338async def test_query_success(handler: DirectConnectionHandler) -> None:
339 """Authorized /user/devices/query should return 200."""
340 req = _make_request(
341 method="POST",
342 headers={**_AUTH_HEADERS, "X-Request-Id": "req-2"},
343 payload={"devices": [{"id": "player1"}]},
344 )
345 mock_result = MagicMock()
346 resp_payload = {"request_id": "req-2", "payload": {"devices": []}}
347 mock_query = patch(
348 "music_assistant.providers.yandex_smarthome.direct.handle_devices_query",
349 new_callable=AsyncMock,
350 return_value=mock_result,
351 )
352 with (
353 mock_query,
354 patch(
355 "music_assistant.providers.yandex_smarthome.direct.asdict", return_value={"devices": []}
356 ),
357 patch(
358 "music_assistant.providers.yandex_smarthome.direct.build_response",
359 return_value=resp_payload,
360 ),
361 ):
362 resp = await handler._handle_query(req)
363 assert resp.status == 200
364
365
366@pytest.mark.asyncio
367async def test_action_success(handler: DirectConnectionHandler) -> None:
368 """Authorized /user/devices/action should return 200."""
369 req = _make_request(
370 method="POST",
371 headers={**_AUTH_HEADERS, "X-Request-Id": "req-3"},
372 payload={"payload": {"devices": []}},
373 )
374 resp_payload = {"request_id": "req-3", "payload": {"devices": []}}
375 mock_action = patch(
376 "music_assistant.providers.yandex_smarthome.direct.handle_devices_action",
377 new_callable=AsyncMock,
378 return_value=MagicMock(),
379 )
380 with (
381 patch(
382 "music_assistant.providers.yandex_smarthome.direct.parse_action_payload",
383 return_value=MagicMock(),
384 ),
385 mock_action,
386 patch(
387 "music_assistant.providers.yandex_smarthome.direct.asdict", return_value={"devices": []}
388 ),
389 patch(
390 "music_assistant.providers.yandex_smarthome.direct.build_response",
391 return_value=resp_payload,
392 ),
393 ):
394 resp = await handler._handle_action(req)
395 assert resp.status == 200
396
397
398@pytest.mark.asyncio
399async def test_unlink_success(handler: DirectConnectionHandler) -> None:
400 """Authorized /user/unlink should return 200."""
401 req = _make_request(
402 method="POST",
403 headers={**_AUTH_HEADERS, "X-Request-Id": "req-4"},
404 )
405 resp_payload = {"request_id": "req-4"}
406 with (
407 patch(
408 "music_assistant.providers.yandex_smarthome.direct.handle_user_unlink",
409 new_callable=AsyncMock,
410 return_value={},
411 ),
412 patch(
413 "music_assistant.providers.yandex_smarthome.direct.build_response",
414 return_value=resp_payload,
415 ),
416 ):
417 resp = await handler._handle_unlink(req)
418 assert resp.status == 200
419
420
421# ---------------------------------------------------------------------------
422# OAuth authorize
423# ---------------------------------------------------------------------------
424
425
426@pytest.mark.asyncio
427async def test_authorize_returns_html(handler: DirectConnectionHandler) -> None:
428 """GET /auth/authorize should return HTML with link button."""
429 req = _make_request(
430 method="GET",
431 path="/auth/authorize",
432 query={
433 "client_id": DIRECT_OAUTH_CLIENT_ID,
434 "redirect_uri": "https://social.yandex.net/broker/redirect",
435 "state": "abc123",
436 "response_type": "code",
437 },
438 )
439 resp = await handler._handle_oauth_authorize(req)
440 assert resp.status == 200
441 assert resp.content_type == "text/html"
442 assert resp.text is not None
443 assert "Music Assistant" in resp.text
444 assert "abc123" in resp.text
445
446
447@pytest.mark.asyncio
448async def test_authorize_missing_redirect_uri(handler: DirectConnectionHandler) -> None:
449 """GET /auth/authorize without redirect_uri should return 400."""
450 req = _make_request(method="GET", path="/auth/authorize", query={})
451 resp = await handler._handle_oauth_authorize(req)
452 assert resp.status == 400
453
454
455@pytest.mark.asyncio
456async def test_authorize_creates_pending_code(handler: DirectConnectionHandler) -> None:
457 """GET /auth/authorize should create a pending authorization code."""
458 req = _make_request(
459 method="GET",
460 path="/auth/authorize",
461 query={
462 "client_id": DIRECT_OAUTH_CLIENT_ID,
463 "redirect_uri": "https://social.yandex.net/broker/redirect",
464 "state": "s1",
465 "response_type": "code",
466 },
467 )
468 assert len(handler._pending_codes) == 0
469 await handler._handle_oauth_authorize(req)
470 assert len(handler._pending_codes) == 1
471
472
473@pytest.mark.asyncio
474async def test_authorize_invalid_client_id(handler: DirectConnectionHandler) -> None:
475 """GET /auth/authorize with wrong client_id should return 400."""
476 req = _make_request(
477 method="GET",
478 path="/auth/authorize",
479 query={
480 "client_id": "wrong-client-id",
481 "redirect_uri": "https://social.yandex.net/broker/redirect",
482 "state": "s1",
483 "response_type": "code",
484 },
485 )
486 resp = await handler._handle_oauth_authorize(req)
487 assert resp.status == 400
488
489
490@pytest.mark.asyncio
491async def test_authorize_invalid_response_type(handler: DirectConnectionHandler) -> None:
492 """GET /auth/authorize with wrong response_type should return 400."""
493 req = _make_request(
494 method="GET",
495 path="/auth/authorize",
496 query={
497 "client_id": DIRECT_OAUTH_CLIENT_ID,
498 "redirect_uri": "https://social.yandex.net/broker/redirect",
499 "state": "s1",
500 "response_type": "token",
501 },
502 )
503 resp = await handler._handle_oauth_authorize(req)
504 assert resp.status == 400
505
506
507@pytest.mark.asyncio
508async def test_authorize_invalid_redirect_uri_domain(handler: DirectConnectionHandler) -> None:
509 """GET /auth/authorize with non-Yandex redirect_uri should return 400."""
510 req = _make_request(
511 method="GET",
512 path="/auth/authorize",
513 query={
514 "client_id": DIRECT_OAUTH_CLIENT_ID,
515 "redirect_uri": "https://evil.example.com/steal",
516 "state": "s1",
517 "response_type": "code",
518 },
519 )
520 resp = await handler._handle_oauth_authorize(req)
521 assert resp.status == 400
522
523
524# ---------------------------------------------------------------------------
525# OAuth token
526# ---------------------------------------------------------------------------
527
528
529@pytest.mark.asyncio
530async def test_token_exchange_valid_code(handler: DirectConnectionHandler) -> None:
531 """Token exchange with valid code should return access_token."""
532 req_auth = _make_request(
533 method="GET",
534 path="/auth/authorize",
535 query={
536 "client_id": DIRECT_OAUTH_CLIENT_ID,
537 "redirect_uri": "https://social.yandex.net/broker/redirect",
538 "state": "s1",
539 "response_type": "code",
540 },
541 )
542 await handler._handle_oauth_authorize(req_auth)
543 code = _get_pending_code(handler)
544
545 req_token = _make_request(
546 method="POST",
547 path="/auth/token",
548 post_data={
549 "grant_type": "authorization_code",
550 "code": code,
551 "client_id": DIRECT_OAUTH_CLIENT_ID,
552 "client_secret": TEST_CLIENT_SECRET,
553 },
554 )
555 resp = await handler._handle_oauth_token(req_token)
556 assert resp.status == 200
557 body = _body_json(resp)
558 assert body["access_token"] == "test-token-abc"
559 assert body["token_type"] == "bearer"
560 assert "refresh_token" in body
561
562
563@pytest.mark.asyncio
564async def test_token_exchange_generates_new_token(
565 handler_no_token: DirectConnectionHandler,
566) -> None:
567 """When no token exists, OAuth should generate a new one."""
568 req_auth = _make_request(
569 method="GET",
570 path="/auth/authorize",
571 query={
572 "client_id": DIRECT_OAUTH_CLIENT_ID,
573 "redirect_uri": "https://social.yandex.net/broker/redirect",
574 "state": "s1",
575 "response_type": "code",
576 },
577 )
578 await handler_no_token._handle_oauth_authorize(req_auth)
579 code = _get_pending_code(handler_no_token)
580
581 req_token = _make_request(
582 method="POST",
583 path="/auth/token",
584 post_data={
585 "grant_type": "authorization_code",
586 "code": code,
587 "client_id": DIRECT_OAUTH_CLIENT_ID,
588 "client_secret": TEST_CLIENT_SECRET,
589 },
590 )
591 resp = await handler_no_token._handle_oauth_token(req_token)
592 assert resp.status == 200
593 body = _body_json(resp)
594 assert body["access_token"]
595 assert len(body["access_token"]) == 32 # uuid4().hex
596 assert len(_handler_no_token_tokens) == 1
597 assert _handler_no_token_tokens[0] == body["access_token"]
598
599
600@pytest.mark.asyncio
601async def test_token_exchange_invalid_client_secret(handler: DirectConnectionHandler) -> None:
602 """Token exchange with wrong client_secret should return 401."""
603 req = _make_request(
604 method="POST",
605 path="/auth/token",
606 post_data={
607 "grant_type": "authorization_code",
608 "code": "any",
609 "client_id": DIRECT_OAUTH_CLIENT_ID,
610 "client_secret": "wrong-secret",
611 },
612 )
613 resp = await handler._handle_oauth_token(req)
614 assert resp.status == 401
615 body = _body_json(resp)
616 assert body["error"] == "invalid_client"
617
618
619@pytest.mark.asyncio
620async def test_token_exchange_invalid_client_id(handler: DirectConnectionHandler) -> None:
621 """Token exchange with wrong client_id should return 401."""
622 req = _make_request(
623 method="POST",
624 path="/auth/token",
625 post_data={
626 "grant_type": "authorization_code",
627 "code": "any",
628 "client_id": "wrong-client-id",
629 "client_secret": TEST_CLIENT_SECRET,
630 },
631 )
632 resp = await handler._handle_oauth_token(req)
633 assert resp.status == 401
634 body = _body_json(resp)
635 assert body["error"] == "invalid_client"
636
637
638@pytest.mark.asyncio
639async def test_token_exchange_invalid_code(handler: DirectConnectionHandler) -> None:
640 """Token exchange with invalid code should return 400."""
641 req = _make_request(
642 method="POST",
643 path="/auth/token",
644 post_data={
645 "grant_type": "authorization_code",
646 "code": "nonexistent",
647 "client_id": DIRECT_OAUTH_CLIENT_ID,
648 "client_secret": TEST_CLIENT_SECRET,
649 },
650 )
651 resp = await handler._handle_oauth_token(req)
652 assert resp.status == 400
653 body = _body_json(resp)
654 assert body["error"] == "invalid_grant"
655
656
657@pytest.mark.asyncio
658async def test_token_exchange_expired_code(handler: DirectConnectionHandler) -> None:
659 """Expired authorization codes should be rejected."""
660 handler._pending_codes["expired-code"] = time.time() - 10
661 req = _make_request(
662 method="POST",
663 path="/auth/token",
664 post_data={
665 "grant_type": "authorization_code",
666 "code": "expired-code",
667 "client_id": DIRECT_OAUTH_CLIENT_ID,
668 "client_secret": TEST_CLIENT_SECRET,
669 },
670 )
671 resp = await handler._handle_oauth_token(req)
672 assert resp.status == 400
673
674
675@pytest.mark.asyncio
676async def test_refresh_token_valid(handler: DirectConnectionHandler) -> None:
677 """Refresh token with correct token should return 200."""
678 req = _make_request(
679 method="POST",
680 path="/auth/token",
681 post_data={
682 "grant_type": "refresh_token",
683 "refresh_token": "test-token-abc",
684 "client_id": DIRECT_OAUTH_CLIENT_ID,
685 "client_secret": TEST_CLIENT_SECRET,
686 },
687 )
688 resp = await handler._handle_oauth_token(req)
689 assert resp.status == 200
690 body = _body_json(resp)
691 assert body["access_token"] == "test-token-abc"
692
693
694@pytest.mark.asyncio
695async def test_refresh_token_invalid(handler: DirectConnectionHandler) -> None:
696 """Refresh token with wrong token should return 400."""
697 req = _make_request(
698 method="POST",
699 path="/auth/token",
700 post_data={
701 "grant_type": "refresh_token",
702 "refresh_token": "wrong",
703 "client_id": DIRECT_OAUTH_CLIENT_ID,
704 "client_secret": TEST_CLIENT_SECRET,
705 },
706 )
707 resp = await handler._handle_oauth_token(req)
708 assert resp.status == 400
709
710
711@pytest.mark.asyncio
712async def test_unsupported_grant_type(handler: DirectConnectionHandler) -> None:
713 """Unsupported grant_type should return 400."""
714 req = _make_request(
715 method="POST",
716 path="/auth/token",
717 post_data={
718 "grant_type": "client_credentials",
719 "client_id": DIRECT_OAUTH_CLIENT_ID,
720 "client_secret": TEST_CLIENT_SECRET,
721 },
722 )
723 resp = await handler._handle_oauth_token(req)
724 assert resp.status == 400
725 body = _body_json(resp)
726 assert body["error"] == "unsupported_grant_type"
727
728
729@pytest.mark.asyncio
730async def test_code_consumed_after_use(handler: DirectConnectionHandler) -> None:
731 """Authorization codes should be single-use."""
732 req_auth = _make_request(
733 method="GET",
734 path="/auth/authorize",
735 query={
736 "client_id": DIRECT_OAUTH_CLIENT_ID,
737 "redirect_uri": "https://social.yandex.net/broker/redirect",
738 "state": "s1",
739 "response_type": "code",
740 },
741 )
742 await handler._handle_oauth_authorize(req_auth)
743 code = _get_pending_code(handler)
744
745 token_post = {
746 "grant_type": "authorization_code",
747 "code": code,
748 "client_id": DIRECT_OAUTH_CLIENT_ID,
749 "client_secret": TEST_CLIENT_SECRET,
750 }
751
752 # First exchange â success
753 req1 = _make_request(method="POST", path="/auth/token", post_data=token_post)
754 resp1 = await handler._handle_oauth_token(req1)
755 assert resp1.status == 200
756
757 # Second exchange â code consumed, should fail
758 req2 = _make_request(method="POST", path="/auth/token", post_data=token_post)
759 resp2 = await handler._handle_oauth_token(req2)
760 assert resp2.status == 400
761
762
763# ---------------------------------------------------------------------------
764# Plugin integration
765# ---------------------------------------------------------------------------
766
767
768def _make_direct_config(**overrides: Any) -> MagicMock:
769 """Create a mock config for direct mode with sensible defaults."""
770 defaults: dict[str, Any] = {
771 "instance_name": "TestMA",
772 "connection_type": CONNECTION_TYPE_DIRECT,
773 "skill_id": "test-skill-id",
774 "skill_token": "test-skill-token",
775 "direct_access_token": "existing-token",
776 "direct_client_secret": TEST_CLIENT_SECRET,
777 "exposed_players": None,
778 "cloud_instance_id": "",
779 "cloud_instance_password": "",
780 "cloud_connection_token": "",
781 "log_level": "GLOBAL",
782 }
783 defaults.update(overrides)
784 config = MagicMock()
785 # accept the optional default arg too: get_setup_value falls through via
786 # get_config_value(key, default), a two-argument call
787 config.get_value = MagicMock(side_effect=lambda key, *_: defaults.get(key, ""))
788 return config
789
790
791@pytest.mark.asyncio
792async def test_start_direct_mode_registers_routes(mock_mass: MagicMock) -> None:
793 """_start_direct_mode should create handler and register routes."""
794 config = _make_direct_config()
795 plugin = YandexSmartHomePlugin(
796 mass=mock_mass,
797 manifest=MagicMock(domain="yandex_smarthome"),
798 config=config,
799 supported_features=set(),
800 )
801 await plugin.handle_async_init()
802 await plugin.loaded_in_mass()
803
804 assert plugin._direct_handler is not None
805 assert mock_mass.webserver.register_dynamic_route.call_count == 10
806 assert plugin._state_notifier is not None
807
808
809@pytest.mark.asyncio
810async def test_start_direct_mode_missing_skill_id(mock_mass: MagicMock) -> None:
811 """
812 Direct mode still registers HTTP routes when skill_id is missing.
813
814 HTTP routes need to be live so Yandex's backend validation during
815 auto-create can succeed; the state notifier is skipped because
816 there is no skill to push state to yet.
817 """
818 config = _make_direct_config(skill_id="")
819 plugin = YandexSmartHomePlugin(
820 mass=mock_mass,
821 manifest=MagicMock(domain="yandex_smarthome"),
822 config=config,
823 supported_features=set(),
824 )
825 plugin.logger = MagicMock()
826 await plugin.handle_async_init()
827 await plugin.loaded_in_mass()
828
829 assert plugin._direct_handler is not None
830 assert mock_mass.webserver.register_dynamic_route.call_count == 10
831 assert plugin._state_notifier is None
832 plugin.logger.info.assert_called()
833
834
835@pytest.mark.asyncio
836async def test_unload_cleans_up_direct(mock_mass: MagicMock) -> None:
837 """unload() should unregister routes and stop notifier."""
838 config = _make_direct_config()
839 plugin = YandexSmartHomePlugin(
840 mass=mock_mass,
841 manifest=MagicMock(domain="yandex_smarthome"),
842 config=config,
843 supported_features=set(),
844 )
845 await plugin.handle_async_init()
846 await plugin.loaded_in_mass()
847 await plugin.unload()
848
849 assert plugin._direct_handler is None
850 assert plugin._state_notifier is None
851