diff --git a/CHANGELOG.md b/CHANGELOG.md index 2f7cc97..d13becb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/). +## [0.12.0] - 2026-04-07 + +### Added + +- `auth_check()` and `auth_refresh()` methods implementing the `Authenticator` protocol — enables `reeln plugins auth google` for credential verification and token renewal + ## [0.11.0] - 2026-04-03 ### Added diff --git a/reeln_google_plugin/__init__.py b/reeln_google_plugin/__init__.py index e7a032a..042c773 100644 --- a/reeln_google_plugin/__init__.py +++ b/reeln_google_plugin/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations -__version__ = "0.11.0" +__version__ = "0.12.0" from reeln_google_plugin.plugin import GooglePlugin diff --git a/reeln_google_plugin/plugin.py b/reeln_google_plugin/plugin.py index 9c6322e..dad7f2a 100644 --- a/reeln_google_plugin/plugin.py +++ b/reeln_google_plugin/plugin.py @@ -6,6 +6,7 @@ from datetime import datetime, timedelta from typing import Any +from reeln.models.auth import AuthCheckResult, AuthStatus from reeln.models.plugin_input import InputField, PluginInputSchema from reeln.models.plugin_schema import ConfigField, PluginConfigSchema from reeln.plugins.hooks import Hook, HookContext @@ -27,7 +28,7 @@ class GooglePlugin: """ name: str = "google" - version: str = "0.11.0" + version: str = "0.12.0" api_version: int = 1 config_schema: PluginConfigSchema = PluginConfigSchema( @@ -567,6 +568,160 @@ def _build_scheduled_start(self, game_info: object) -> str | None: log.warning("Google plugin: could not parse game time '%s %s', using default", date_str, game_time) return None + # ------------------------------------------------------------------ + # Authenticator capability + # ------------------------------------------------------------------ + + def auth_check(self) -> list[AuthCheckResult]: + """Test Google/YouTube authentication and return check results.""" + from pathlib import Path + + client_secrets = self._config.get("client_secrets_file") + if not client_secrets: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.NOT_CONFIGURED, + message="client_secrets_file not configured", + hint="Set client_secrets_file in plugin config", + ) + ] + + credentials_cache_str = self._config.get("credentials_cache") + credentials_cache = ( + Path(credentials_cache_str) + if credentials_cache_str + else auth.default_credentials_path() + ) + scopes = self._config.get("scopes") + + try: + creds = auth.get_credentials( + Path(client_secrets), + credentials_cache, + scopes=scopes, + ) + except auth.AuthError as exc: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.FAIL, + message=str(exc), + hint="Run 'reeln plugins auth --refresh google' to re-authenticate", + ) + ] + + try: + youtube = auth.build_youtube_service(creds) + except auth.AuthError as exc: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.FAIL, + message=str(exc), + ) + ] + + return self._check_channel(youtube, creds) + + def auth_refresh(self) -> list[AuthCheckResult]: + """Clear cached credentials and re-authenticate.""" + from pathlib import Path + + client_secrets = self._config.get("client_secrets_file") + if not client_secrets: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.NOT_CONFIGURED, + message="client_secrets_file not configured", + hint="Set client_secrets_file in plugin config", + ) + ] + + credentials_cache_str = self._config.get("credentials_cache") + credentials_cache = ( + Path(credentials_cache_str) + if credentials_cache_str + else auth.default_credentials_path() + ) + scopes = self._config.get("scopes") + + try: + creds = auth.get_credentials( + Path(client_secrets), + credentials_cache, + scopes=scopes, + fresh=True, + ) + except auth.AuthError as exc: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.FAIL, + message=str(exc), + ) + ] + + try: + youtube = auth.build_youtube_service(creds) + except auth.AuthError as exc: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.FAIL, + message=str(exc), + ) + ] + + self._youtube = youtube + return self._check_channel(youtube, creds) + + def _check_channel(self, youtube: Any, creds: Any) -> list[AuthCheckResult]: + """Query the authenticated channel and return an auth result.""" + granted_scopes: list[str] = [] + if hasattr(creds, "scopes") and creds.scopes: + granted_scopes = sorted(creds.scopes) + + try: + resp = youtube.channels().list(part="snippet", mine=True).execute() + except Exception as exc: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.FAIL, + message=f"channels.list failed: {exc}", + scopes=granted_scopes, + required_scopes=sorted(auth.DEFAULT_SCOPES), + ) + ] + + items = resp.get("items", []) + if not items: + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.WARN, + message="No channel found for authenticated user", + scopes=granted_scopes, + required_scopes=sorted(auth.DEFAULT_SCOPES), + ) + ] + + channel = items[0] + title = channel.get("snippet", {}).get("title", "") + + return [ + AuthCheckResult( + service="YouTube", + status=AuthStatus.OK, + message="Authenticated", + identity=title, + scopes=granted_scopes, + required_scopes=sorted(auth.DEFAULT_SCOPES), + ) + ] + def _build_title(self, game_info: object) -> str: """Build a livestream title from game info.""" home_team = getattr(game_info, "home_team", "") diff --git a/tests/unit/test_plugin.py b/tests/unit/test_plugin.py index 1dd856d..ad3e1a6 100644 --- a/tests/unit/test_plugin.py +++ b/tests/unit/test_plugin.py @@ -9,10 +9,11 @@ from unittest.mock import MagicMock, patch import pytest +from reeln.models.auth import AuthStatus from reeln.plugins.hooks import Hook, HookContext from reeln.plugins.registry import HookRegistry -from reeln_google_plugin.auth import AuthError +from reeln_google_plugin.auth import DEFAULT_SCOPES, AuthError from reeln_google_plugin.livestream import LivestreamError from reeln_google_plugin.playlist import PlaylistError from reeln_google_plugin.plugin import GooglePlugin @@ -2463,3 +2464,260 @@ def test_fetch_failure_non_fatal( # State still reset assert plugin._game_info is None assert plugin._youtube is None + + +class TestAuthCheck: + """Tests for GooglePlugin.auth_check().""" + + def test_not_configured(self) -> None: + """client_secrets_file missing -> NOT_CONFIGURED.""" + plugin = GooglePlugin() + results = plugin.auth_check() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.NOT_CONFIGURED + assert "client_secrets_file" in r.message + assert r.hint + + @patch("reeln_google_plugin.plugin.auth") + def test_auth_error( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """get_credentials raises AuthError -> FAIL.""" + mock_auth.AuthError = AuthError + mock_auth.get_credentials.side_effect = AuthError("token revoked") + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_check() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.FAIL + assert "token revoked" in r.message + assert "refresh" in r.hint.lower() + + @patch("reeln_google_plugin.plugin.auth") + def test_build_service_error( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """build_youtube_service raises AuthError -> FAIL.""" + mock_auth.AuthError = AuthError + mock_auth.get_credentials.return_value = MagicMock() + mock_auth.build_youtube_service.side_effect = AuthError("service build failed") + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_check() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.FAIL + assert "service build failed" in r.message + + @patch("reeln_google_plugin.plugin.auth") + def test_channels_list_fails( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """API call raises exception -> FAIL with scopes.""" + mock_auth.AuthError = AuthError + fake_creds = MagicMock() + fake_creds.scopes = {"https://www.googleapis.com/auth/youtube"} + mock_auth.get_credentials.return_value = fake_creds + mock_auth.DEFAULT_SCOPES = DEFAULT_SCOPES + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + mock_youtube = MagicMock() + mock_youtube.channels().list().execute.side_effect = Exception("403 forbidden") + mock_auth.build_youtube_service.return_value = mock_youtube + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_check() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.FAIL + assert "channels.list failed" in r.message + assert "403 forbidden" in r.message + assert r.scopes + assert r.required_scopes + + @patch("reeln_google_plugin.plugin.auth") + def test_no_channel_found( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """channels.list returns empty items -> WARN.""" + mock_auth.AuthError = AuthError + fake_creds = MagicMock() + fake_creds.scopes = set(DEFAULT_SCOPES) + mock_auth.get_credentials.return_value = fake_creds + mock_auth.DEFAULT_SCOPES = DEFAULT_SCOPES + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + mock_youtube = MagicMock() + mock_youtube.channels().list().execute.return_value = {"items": []} + mock_auth.build_youtube_service.return_value = mock_youtube + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_check() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.WARN + assert "No channel found" in r.message + assert r.scopes + assert r.required_scopes + + @patch("reeln_google_plugin.plugin.auth") + def test_success( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """channels.list returns channel with title -> OK with identity.""" + mock_auth.AuthError = AuthError + fake_creds = MagicMock() + fake_creds.scopes = set(DEFAULT_SCOPES) + mock_auth.get_credentials.return_value = fake_creds + mock_auth.DEFAULT_SCOPES = DEFAULT_SCOPES + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + mock_youtube = MagicMock() + mock_youtube.channels().list().execute.return_value = { + "items": [{"snippet": {"title": "StreamnDad"}}] + } + mock_auth.build_youtube_service.return_value = mock_youtube + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_check() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.OK + assert r.message == "Authenticated" + assert r.identity == "StreamnDad" + assert r.scopes + assert r.required_scopes + + +class TestAuthRefresh: + """Tests for GooglePlugin.auth_refresh().""" + + def test_not_configured(self) -> None: + """client_secrets_file missing -> NOT_CONFIGURED.""" + plugin = GooglePlugin() + results = plugin.auth_refresh() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.NOT_CONFIGURED + assert "client_secrets_file" in r.message + assert r.hint + + @patch("reeln_google_plugin.plugin.auth") + def test_success( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """fresh=True passed, channel found -> OK.""" + mock_auth.AuthError = AuthError + fake_creds = MagicMock() + fake_creds.scopes = set(DEFAULT_SCOPES) + mock_auth.get_credentials.return_value = fake_creds + mock_auth.DEFAULT_SCOPES = DEFAULT_SCOPES + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + mock_youtube = MagicMock() + mock_youtube.channels().list().execute.return_value = { + "items": [{"snippet": {"title": "StreamnDad"}}] + } + mock_auth.build_youtube_service.return_value = mock_youtube + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_refresh() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.OK + assert r.identity == "StreamnDad" + + # Verify fresh=True was passed + call_kwargs = mock_auth.get_credentials.call_args + assert call_kwargs.kwargs.get("fresh") is True + + # Verify youtube service cached on plugin + assert plugin._youtube is mock_youtube + + @patch("reeln_google_plugin.plugin.auth") + def test_auth_error( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """get_credentials(fresh=True) raises -> FAIL.""" + mock_auth.AuthError = AuthError + mock_auth.get_credentials.side_effect = AuthError("consent flow cancelled") + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_refresh() + + assert len(results) == 1 + r = results[0] + assert r.service == "YouTube" + assert r.status == AuthStatus.FAIL + assert "consent flow cancelled" in r.message + + # Verify fresh=True was passed even though it failed + call_kwargs = mock_auth.get_credentials.call_args + assert call_kwargs.kwargs.get("fresh") is True + + @patch("reeln_google_plugin.plugin.auth") + def test_build_service_error( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """build_youtube_service raises during refresh -> FAIL.""" + mock_auth.AuthError = AuthError + fake_creds = MagicMock() + mock_auth.get_credentials.return_value = fake_creds + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + mock_auth.build_youtube_service.side_effect = AuthError("missing lib") + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_refresh() + + assert len(results) == 1 + assert results[0].status == AuthStatus.FAIL + assert "missing lib" in results[0].message + + +class TestCheckChannelNoScopes: + """Test _check_channel when creds have no scopes attribute.""" + + @patch("reeln_google_plugin.plugin.auth") + def test_creds_without_scopes( + self, mock_auth: MagicMock, plugin_config: dict[str, Any] + ) -> None: + """Creds without scopes attribute still return results.""" + mock_auth.AuthError = AuthError + fake_creds = MagicMock(spec=[]) # no scopes attribute + mock_auth.get_credentials.return_value = fake_creds + mock_auth.DEFAULT_SCOPES = DEFAULT_SCOPES + mock_auth.default_credentials_path.return_value = Path("/tmp/oauth.json") + + mock_youtube = MagicMock() + mock_youtube.channels().list().execute.return_value = { + "items": [{"snippet": {"title": "TestChannel"}}] + } + mock_auth.build_youtube_service.return_value = mock_youtube + + plugin = GooglePlugin(plugin_config) + results = plugin.auth_check() + + assert len(results) == 1 + assert results[0].status == AuthStatus.OK + assert results[0].scopes == [] # no scopes on creds