|
| 1 | +import time |
| 2 | +import streamlit as st |
| 3 | +import pytest |
| 4 | +from unittest.mock import AsyncMock |
| 5 | + |
| 6 | +from streamlit_oauth import OAuth2Component, OAuth2, StreamlitOauthError |
| 7 | + |
| 8 | + |
| 9 | +def test_authorize_button_success(monkeypatch): |
| 10 | + st.session_state.clear() |
| 11 | + client = OAuth2("id", "secret", "auth", "token") |
| 12 | + oauth = OAuth2Component(client=client) |
| 13 | + |
| 14 | + # Mock async client methods |
| 15 | + monkeypatch.setattr(oauth.client, "get_authorization_url", AsyncMock(return_value="http://auth")) |
| 16 | + monkeypatch.setattr(oauth.client, "get_access_token", AsyncMock(return_value={"access_token": "tok"})) |
| 17 | + |
| 18 | + # Force deterministic state and component output |
| 19 | + monkeypatch.setattr("streamlit_oauth._generate_state", lambda key=None: "STATE") |
| 20 | + monkeypatch.setattr("streamlit_oauth._authorize_button", lambda **kwargs: {"code": "CODE", "state": "STATE"}) |
| 21 | + |
| 22 | + result = oauth.authorize_button("Login", "http://cb", "scope", key="k") |
| 23 | + assert result["token"]["access_token"] == "tok" |
| 24 | + assert f"state-k" not in st.session_state |
| 25 | + |
| 26 | + |
| 27 | +def test_authorize_button_state_mismatch(monkeypatch): |
| 28 | + st.session_state.clear() |
| 29 | + client = OAuth2("id", "secret", "auth", "token") |
| 30 | + oauth = OAuth2Component(client=client) |
| 31 | + |
| 32 | + monkeypatch.setattr(oauth.client, "get_authorization_url", AsyncMock(return_value="http://auth")) |
| 33 | + monkeypatch.setattr(oauth.client, "get_access_token", AsyncMock(return_value={"access_token": "tok"})) |
| 34 | + monkeypatch.setattr("streamlit_oauth._generate_state", lambda key=None: "GOOD") |
| 35 | + monkeypatch.setattr("streamlit_oauth._authorize_button", lambda **kwargs: {"code": "CODE", "state": "BAD"}) |
| 36 | + |
| 37 | + with pytest.raises(StreamlitOauthError): |
| 38 | + oauth.authorize_button("Login", "http://cb", "scope", key="k") |
| 39 | + |
| 40 | + |
| 41 | +def test_refresh_token_expired(monkeypatch): |
| 42 | + client = OAuth2("id", "secret", "auth", "token") |
| 43 | + oauth = OAuth2Component(client=client) |
| 44 | + |
| 45 | + monkeypatch.setattr(oauth.client, "refresh_token", AsyncMock(return_value={"access_token": "new"})) |
| 46 | + |
| 47 | + token = {"access_token": "old", "refresh_token": "r", "expires_at": time.time() - 1} |
| 48 | + result = oauth.refresh_token(token) |
| 49 | + |
| 50 | + assert result["access_token"] == "new" |
| 51 | + |
| 52 | + |
| 53 | +def test_revoke_token(monkeypatch): |
| 54 | + client = OAuth2("id", "secret", "auth", "token") |
| 55 | + oauth = OAuth2Component(client=client) |
| 56 | + revoke_mock = AsyncMock() |
| 57 | + monkeypatch.setattr(oauth.client, "revoke_token", revoke_mock) |
| 58 | + |
| 59 | + assert oauth.revoke_token({"access_token": "abc"}) is True |
| 60 | + revoke_mock.assert_awaited_once() |
0 commit comments