80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
|
import json
|
||
|
|
import time
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from pigway_frame_flow.errors import FlowError
|
||
|
|
from pigway_frame_flow.providers.oauth import OAuthTokens
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("kind", ["google-drive", "alipan", "baidu"])
|
||
|
|
def test_refresh_fixed_endpoint_private_persistence(
|
||
|
|
service, monkeypatch, tmp_path, kind
|
||
|
|
):
|
||
|
|
calls = []
|
||
|
|
|
||
|
|
def fetch(url, **kwargs):
|
||
|
|
calls.append((url, kwargs))
|
||
|
|
return json.dumps(
|
||
|
|
{
|
||
|
|
"access_token": "new-private-token",
|
||
|
|
"refresh_token": "rotated-private-token",
|
||
|
|
"expires_in": 3600,
|
||
|
|
}
|
||
|
|
).encode()
|
||
|
|
|
||
|
|
monkeypatch.setattr(service.assets, "fetch_url", fetch)
|
||
|
|
credentials = {
|
||
|
|
"access_token": "old-token",
|
||
|
|
"refresh_token": "refresh-private",
|
||
|
|
"client_id": "app",
|
||
|
|
"client_secret": "client-private",
|
||
|
|
}
|
||
|
|
path = tmp_path / "credentials.secret"
|
||
|
|
result = OAuthTokens(service.settings, service.assets).refresh(
|
||
|
|
kind, credentials, path
|
||
|
|
)
|
||
|
|
assert calls[0][0] == OAuthTokens.ENDPOINTS[kind]
|
||
|
|
assert result["expires_at"] > time.time() + 3500
|
||
|
|
assert json.loads(path.read_text())["refresh_token"] == "rotated-private-token"
|
||
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
||
|
|
assert service.settings.sanitize("new-private-token") != "new-private-token"
|
||
|
|
|
||
|
|
|
||
|
|
def test_refresh_failure_never_retries_or_exposes_secret(service, monkeypatch):
|
||
|
|
calls = []
|
||
|
|
|
||
|
|
def fetch(*args, **kwargs):
|
||
|
|
calls.append(args)
|
||
|
|
raise RuntimeError("private-secret")
|
||
|
|
|
||
|
|
monkeypatch.setattr(service.assets, "fetch_url", fetch)
|
||
|
|
with pytest.raises(FlowError) as exc:
|
||
|
|
OAuthTokens(service.settings, service.assets).refresh(
|
||
|
|
"baidu",
|
||
|
|
{
|
||
|
|
"refresh_token": "refresh",
|
||
|
|
"client_id": "app",
|
||
|
|
"client_secret": "private-secret",
|
||
|
|
},
|
||
|
|
)
|
||
|
|
assert len(calls) == 1
|
||
|
|
assert "private-secret" not in str(exc.value)
|
||
|
|
|
||
|
|
|
||
|
|
def test_invalid_refresh_does_not_replace_existing_credentials(
|
||
|
|
service, monkeypatch, tmp_path
|
||
|
|
):
|
||
|
|
path = tmp_path / "secret"
|
||
|
|
path.write_text("original")
|
||
|
|
monkeypatch.setattr(
|
||
|
|
service.assets, "fetch_url", lambda *a, **k: b'{"error":"invalid_grant"}'
|
||
|
|
)
|
||
|
|
with pytest.raises(FlowError):
|
||
|
|
OAuthTokens(service.settings, service.assets).refresh(
|
||
|
|
"google-drive",
|
||
|
|
{"refresh_token": "refresh", "client_id": "app", "client_secret": "secret"},
|
||
|
|
path,
|
||
|
|
)
|
||
|
|
assert path.read_text() == "original"
|