You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
# backend/tests/test_oauth_mock.py — C3 OAuth 플로우 + C8 토큰 만료
|
|
# (respx 전역상태 플레이크 회피 위해 httpx.post 를 직접 monkeypatch — 결정적)
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from app.config import get_settings
|
|
from app.connectors import oauth
|
|
from app.crypto import decrypt_token, encrypt_token
|
|
from app.models import ConnectorAccount, ConnectorDomain, ConnState
|
|
|
|
|
|
class _FakeResp:
|
|
def __init__(self, status, payload=None):
|
|
self.status_code = status
|
|
self._payload = payload or {}
|
|
|
|
def json(self):
|
|
return self._payload
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise RuntimeError(f"http {self.status_code}")
|
|
|
|
|
|
@pytest.fixture()
|
|
def google_creds(monkeypatch):
|
|
monkeypatch.setenv("GOOGLE_CLIENT_ID", "cid")
|
|
monkeypatch.setenv("GOOGLE_CLIENT_SECRET", "sec")
|
|
get_settings.cache_clear()
|
|
yield
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_oauth_callback_stores_encrypted_token(session, google_creds, monkeypatch):
|
|
s, _ = session
|
|
url = oauth.start_oauth(s, "mail", "gmail")
|
|
assert "code_challenge=" in url and "state=" in url # PKCE + state
|
|
state = url.split("state=")[1].split("&")[0]
|
|
|
|
# 평문 노출 검사가 결정적이도록 충분히 distinctive 한 토큰 사용(짧은 'AT'는 base64에 우연히 등장).
|
|
access = "ACCESS-TOKEN-Zx9Q7w"
|
|
monkeypatch.setattr(
|
|
oauth.httpx,
|
|
"post",
|
|
lambda *a, **k: _FakeResp(
|
|
200,
|
|
{
|
|
"access_token": access,
|
|
"refresh_token": "RT",
|
|
"expires_in": 3600,
|
|
"scope": "https://www.googleapis.com/auth/gmail.readonly",
|
|
},
|
|
),
|
|
)
|
|
# _fetch_identity 의 실 네트워크 호출 차단(이메일 없음 → 단일 계정 폴백).
|
|
monkeypatch.setattr(oauth.httpx, "get", lambda *a, **k: _FakeResp(200, {"emailAddress": ""}))
|
|
acct = oauth.finish_oauth(s, code="CODE", state=state)
|
|
assert acct.mode == "real" and acct.state == ConnState.connected
|
|
assert acct.token_enc and access not in acct.token_enc # 평문 노출 금지
|
|
assert decrypt_token(acct.token_enc)["access_token"] == access
|
|
from app.models import OAuthState
|
|
|
|
assert s.get(OAuthState, state) is None # state row 소거
|
|
|
|
|
|
def test_token_refresh_failure_marks_expired(session, google_creds, monkeypatch):
|
|
s, _ = session
|
|
acct = ConnectorAccount(
|
|
id="ca-mail-gmail",
|
|
domain=ConnectorDomain.mail,
|
|
provider="gmail",
|
|
token_enc=encrypt_token(
|
|
{"access_token": "old", "refresh_token": "RT", "expires_at": int(time.time()) - 10}
|
|
),
|
|
)
|
|
s.add(acct)
|
|
s.commit()
|
|
monkeypatch.setattr(oauth.httpx, "post", lambda *a, **k: _FakeResp(401))
|
|
with pytest.raises(PermissionError):
|
|
oauth.valid_access_token(s, acct)
|
|
assert acct.state == ConnState.token_expired
|
|
|
|
|
|
def test_token_valid_when_unexpired(session, google_creds):
|
|
s, _ = session
|
|
acct = ConnectorAccount(
|
|
id="ca-x",
|
|
domain=ConnectorDomain.mail,
|
|
provider="gmail",
|
|
token_enc=encrypt_token({"access_token": "FRESH", "expires_at": int(time.time()) + 9999}),
|
|
)
|
|
assert oauth.valid_access_token(s, acct) == "FRESH"
|