-
Notifications
You must be signed in to change notification settings - Fork 22
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
mob next [ci-skip] [ci skip] [skip ci]
lastFile:skore/tests/unit/hub/test_login.py
- Loading branch information
1 parent
75734a5
commit 993d9f7
Showing
5 changed files
with
140 additions
and
130 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,134 @@ | ||
| import time | ||
| from threading import Thread | ||
| from urllib.parse import urljoin | ||
|
|
||
| import httpx | ||
| import pytest | ||
| from httpx import Response | ||
| from skore.hub.api import URI | ||
| from skore.hub.client import AuthenticationError | ||
| from skore.hub.login import login | ||
|
|
||
|
|
||
| CALLBACK_URL = urljoin(URI, "identity/oauth/device/callback") | ||
| LOGIN_URL = urljoin(URI, "identity/oauth/device/login") | ||
| TOKEN_URL = urljoin(URI, "identity/oauth/device/token") | ||
|
|
||
|
|
||
| @pytest.mark.respx(assert_all_called=True) | ||
| def test_manual_login(monkeypatch, respx_mock, mock_now, mock_nowstr): | ||
| def webbrowser_open(url): | ||
| webbrowser_open.url = url | ||
|
|
||
| monkeypatch.setattr("webbrowser.open", webbrowser_open) | ||
|
|
||
| respx_mock.get(LOGIN_URL).mock( | ||
| Response( | ||
| 200, | ||
| json={ | ||
| "authorization_url": "url", | ||
| "device_code": "device", | ||
| "user_code": "user", | ||
| }, | ||
| ) | ||
| ) | ||
| respx_mock.get(TOKEN_URL).mock( | ||
| Response( | ||
| 200, | ||
| json={ | ||
| "token": { | ||
| "access_token": "A", | ||
| "refresh_token": "B", | ||
| "expires_at": mock_nowstr, | ||
| } | ||
| }, | ||
| ) | ||
| ) | ||
|
|
||
| token = login(auto_otp=False) | ||
|
|
||
| assert webbrowser_open.url == "url" | ||
| assert token.access == "A" | ||
| assert token.refreshment == "B" | ||
| assert token.expires_at == mock_now | ||
|
|
||
|
|
||
| @pytest.mark.respx(assert_all_called=True) | ||
| def test_manual_login_timeout(monkeypatch, respx_mock): | ||
| monkeypatch.setattr("webbrowser.open", lambda _: None) | ||
|
|
||
| respx_mock.get(TOKEN_URL).mock(Response(500)) | ||
| respx_mock.get(LOGIN_URL).mock( | ||
| Response( | ||
| 200, | ||
| json={ | ||
| "authorization_url": "https://idp.com", | ||
| "device_code": "device-123", | ||
| "user_code": "123", | ||
| }, | ||
| ) | ||
| ) | ||
|
|
||
| with pytest.raises(AuthenticationError, match="Timeout"): | ||
| login(timeout=0, auto_otp=False) | ||
|
|
||
|
|
||
| @pytest.mark.respx(assert_all_called=True) | ||
| def test_auto_otp_login(monkeypatch, respx_mock, mock_now, mock_nowstr): | ||
| def webbrowser_open(url): | ||
| webbrowser_open.url = url | ||
|
|
||
| monkeypatch.setattr("webbrowser.open", webbrowser_open) | ||
| respx_mock.route(host="localhost").pass_through() | ||
|
|
||
| login_route = respx_mock.get(LOGIN_URL).mock( | ||
| Response( | ||
| 200, | ||
| json={ | ||
| "authorization_url": "url", | ||
| "device_code": "device", | ||
| "user_code": "user", | ||
| }, | ||
| ) | ||
| ) | ||
| respx_mock.post(CALLBACK_URL).mock(Response(201)) | ||
| respx_mock.get(TOKEN_URL).mock( | ||
| Response( | ||
| 200, | ||
| json={ | ||
| "token": { | ||
| "access_token": "A", | ||
| "refresh_token": "B", | ||
| "expires_at": mock_nowstr, | ||
| } | ||
| }, | ||
| ) | ||
| ) | ||
|
|
||
| port = -1 | ||
|
|
||
| def mock_create_server(callback): | ||
| from skore.hub.callback_server import launch_callback_server | ||
|
|
||
| nonlocal port | ||
| port = launch_callback_server(callback) | ||
| return port | ||
|
|
||
| monkeypatch.setattr("skore.hub.login.launch_callback_server", mock_create_server) | ||
|
|
||
| def call_success(): | ||
| with httpx.Client() as client: | ||
| while not login_route.called: | ||
| time.sleep(0.5) | ||
|
|
||
| client.get(f"http://localhost:{port}") | ||
|
|
||
| success_thread = Thread(target=call_success) | ||
| success_thread.start() | ||
|
|
||
| token = login(auto_otp=True) | ||
|
|
||
| assert webbrowser_open.url == "url" | ||
| assert token.access == "A" | ||
| assert token.refreshment == "B" | ||
| assert token.expires_at == mock_now |