fde9a439df
Backend: cache Google access tokens (drop dead access_token_expires_at, migration 0007), handle MeTube cleared/canceled events by URL, return email from /auth/status only when authenticated, move Google base URLs into settings, run container as non-root. Frontend: include local feed filters in the query key, remove dead Saved page and unused assets, drop stale CategoryNav props and classes, send Content-Type only with a body, remove Uncategorized from sidebar.
226 lines
7.6 KiB
Python
226 lines
7.6 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api import auth as auth_module
|
|
from app.db import get_db
|
|
from app.main import app
|
|
from app.models.oauth_credentials import SINGLETON_ID, OAuthCredentials
|
|
from app.services import google_oauth
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch, db_session):
|
|
def _get_db_override():
|
|
yield db_session
|
|
|
|
app.dependency_overrides[get_db] = _get_db_override
|
|
monkeypatch.setattr(auth_module, "_run_initial_sync", lambda: None)
|
|
monkeypatch.setattr(
|
|
google_oauth, "build_authorization_url", lambda: ("https://accounts.google.com/fake", "fixed-state")
|
|
)
|
|
# Deliberately not using `with TestClient(app)`: that runs the app's
|
|
# lifespan, which would try to reach the real MeTube instance and DB.
|
|
yield TestClient(app)
|
|
del app.dependency_overrides[get_db]
|
|
|
|
|
|
class FakeCredentials:
|
|
token = "fake-access-token"
|
|
refresh_token = "fake-refresh-token"
|
|
expiry = None
|
|
|
|
|
|
def _start_oauth(client):
|
|
response = client.get("/api/auth/google/start", follow_redirects=False)
|
|
assert response.status_code in (302, 307)
|
|
return "fixed-state"
|
|
|
|
|
|
def test_callback_rejects_disallowed_email(client, monkeypatch):
|
|
state = _start_oauth(client)
|
|
|
|
monkeypatch.setattr(google_oauth, "exchange_code", lambda code, state: FakeCredentials())
|
|
monkeypatch.setattr(google_oauth, "fetch_userinfo", lambda token: {"email": "stranger@example.com"})
|
|
revoked = {}
|
|
monkeypatch.setattr(google_oauth, "revoke_token", lambda token: revoked.setdefault("called", True))
|
|
|
|
response = client.get(f"/api/auth/google/callback?code=abc&state={state}", follow_redirects=False)
|
|
|
|
assert response.status_code in (302, 307)
|
|
assert "account_not_allowed" in response.headers["location"]
|
|
assert revoked.get("called") is True
|
|
|
|
status = client.get("/api/auth/status").json()
|
|
assert status["authenticated"] is False
|
|
assert status["connected"] is False
|
|
|
|
|
|
def test_callback_accepts_allowed_email(client, monkeypatch):
|
|
state = _start_oauth(client)
|
|
|
|
monkeypatch.setattr(google_oauth, "exchange_code", lambda code, state: FakeCredentials())
|
|
monkeypatch.setattr(google_oauth, "fetch_userinfo", lambda token: {"email": "owner@example.com"})
|
|
|
|
response = client.get(f"/api/auth/google/callback?code=abc&state={state}", follow_redirects=False)
|
|
|
|
assert response.status_code in (302, 307)
|
|
assert "auth_error" not in response.headers["location"]
|
|
|
|
status = client.get("/api/auth/status").json()
|
|
assert status["authenticated"] is True
|
|
assert status["connected"] is True
|
|
assert status["email"] == "owner@example.com"
|
|
|
|
|
|
def test_callback_rejects_invalid_state(client, monkeypatch):
|
|
_start_oauth(client)
|
|
|
|
response = client.get("/api/auth/google/callback?code=abc&state=wrong-state", follow_redirects=False)
|
|
|
|
assert response.status_code in (302, 307)
|
|
assert "invalid_state" in response.headers["location"]
|
|
|
|
|
|
def test_status_hides_email_without_session(client, db_session):
|
|
"""/api/auth/status is public (the Connect page reads `authenticated`), so
|
|
the connected account's email must not leak to unauthenticated requests."""
|
|
row = OAuthCredentials(
|
|
id=SINGLETON_ID,
|
|
google_email="owner@example.com",
|
|
encrypted_refresh_token="encrypted-blob",
|
|
)
|
|
db_session.add(row)
|
|
db_session.commit()
|
|
|
|
status = client.get("/api/auth/status").json()
|
|
|
|
assert status["authenticated"] is False
|
|
assert status["connected"] is True
|
|
assert status["email"] is None
|
|
|
|
|
|
def _seed_credentials_row(db_session, refresh_token="refresh-1"):
|
|
row = OAuthCredentials(
|
|
id=SINGLETON_ID,
|
|
google_email="owner@example.com",
|
|
encrypted_refresh_token=google_oauth.encrypt_token(refresh_token),
|
|
)
|
|
db_session.add(row)
|
|
db_session.commit()
|
|
|
|
|
|
class FakeGoogleCredentials:
|
|
"""Stands in for google.oauth2.credentials.Credentials in cache tests."""
|
|
|
|
def __init__(self, **kwargs):
|
|
self.refresh_token = kwargs["refresh_token"]
|
|
self.token = None
|
|
self.expiry = None
|
|
|
|
def refresh(self, request):
|
|
raise NotImplementedError
|
|
|
|
|
|
def test_get_credentials_caches_and_reuses_access_token(monkeypatch, db_session):
|
|
monkeypatch.setattr(google_oauth, "_cached_credentials", None)
|
|
_seed_credentials_row(db_session)
|
|
|
|
refreshes = []
|
|
|
|
class Creds(FakeGoogleCredentials):
|
|
def refresh(self, request):
|
|
refreshes.append(request)
|
|
self.token = "access-token"
|
|
self.expiry = datetime(2099, 1, 1, tzinfo=timezone.utc)
|
|
|
|
monkeypatch.setattr(google_oauth, "Credentials", Creds)
|
|
|
|
first = google_oauth.get_credentials(db_session)
|
|
second = google_oauth.get_credentials(db_session)
|
|
|
|
assert first is second
|
|
assert first.token == "access-token"
|
|
assert len(refreshes) == 1 # second call must not hit Google again
|
|
|
|
|
|
def test_get_credentials_refreshes_expiring_token(monkeypatch, db_session):
|
|
monkeypatch.setattr(google_oauth, "_cached_credentials", None)
|
|
_seed_credentials_row(db_session)
|
|
|
|
refreshes = []
|
|
|
|
class Creds(FakeGoogleCredentials):
|
|
def refresh(self, request):
|
|
refreshes.append(request)
|
|
self.token = "access-token"
|
|
# Expires in 30s -- inside the 60s refresh buffer.
|
|
self.expiry = datetime.now(timezone.utc) + timedelta(seconds=30)
|
|
|
|
monkeypatch.setattr(google_oauth, "Credentials", Creds)
|
|
|
|
google_oauth.get_credentials(db_session)
|
|
google_oauth.get_credentials(db_session)
|
|
|
|
assert len(refreshes) == 2
|
|
|
|
|
|
def test_get_credentials_refresh_failure_clears_cache(monkeypatch, db_session):
|
|
"""A failed token refresh must not leave a broken cached object behind:
|
|
the error propagates and the next call rebuilds credentials from the DB
|
|
instead of reusing (and re-failing on) the stale cache entry."""
|
|
monkeypatch.setattr(google_oauth, "_cached_credentials", None)
|
|
_seed_credentials_row(db_session)
|
|
|
|
built = []
|
|
|
|
class Creds(FakeGoogleCredentials):
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
built.append(self.refresh_token)
|
|
|
|
def refresh(self, request):
|
|
raise RuntimeError("token endpoint down")
|
|
|
|
monkeypatch.setattr(google_oauth, "Credentials", Creds)
|
|
|
|
for _ in range(2):
|
|
with pytest.raises(RuntimeError, match="token endpoint down"):
|
|
google_oauth.get_credentials(db_session)
|
|
|
|
# Both calls failed, and the broken cache entry was dropped: each call
|
|
# rebuilt from the stored refresh token instead of sticking.
|
|
assert len(built) == 2
|
|
assert google_oauth._cached_credentials is None
|
|
|
|
|
|
def test_get_credentials_rebuilds_cache_after_reconnect(monkeypatch, db_session):
|
|
monkeypatch.setattr(google_oauth, "_cached_credentials", None)
|
|
_seed_credentials_row(db_session)
|
|
|
|
built = []
|
|
|
|
class Creds(FakeGoogleCredentials):
|
|
def __init__(self, **kwargs):
|
|
super().__init__(**kwargs)
|
|
built.append(self.refresh_token)
|
|
self.token = "already-valid"
|
|
self.expiry = datetime(2099, 1, 1, tzinfo=timezone.utc)
|
|
|
|
def refresh(self, request):
|
|
raise AssertionError("must not refresh a still-valid cached token")
|
|
|
|
monkeypatch.setattr(google_oauth, "Credentials", Creds)
|
|
|
|
google_oauth.get_credentials(db_session)
|
|
assert built == ["refresh-1"]
|
|
|
|
# Reconnect: the stored refresh token changes, the old cache entry must
|
|
# not be reused.
|
|
row = db_session.get(OAuthCredentials, SINGLETON_ID)
|
|
row.encrypted_refresh_token = google_oauth.encrypt_token("refresh-2")
|
|
db_session.commit()
|
|
|
|
google_oauth.get_credentials(db_session)
|
|
assert built == ["refresh-1", "refresh-2"]
|