test_auth_oauth_helpers.py
1 """Tests for auth_setup pure OAuth helpers. 2 3 These helpers are the shared building blocks between the interactive 4 CLI (``setup_google_ads``) and the forthcoming web-based auth wizard. 5 Extracting them so both paths build the same Flow / auth-URL shape 6 prevents drift. 7 """ 8 9 from __future__ import annotations 10 11 from unittest.mock import MagicMock, patch 12 13 import pytest 14 15 # Module imports under test — some of these don't exist yet. 16 from mureo.auth_setup import ( # noqa: I001 17 build_google_client_config, 18 build_google_flow, 19 exchange_google_code, 20 google_auth_url, 21 ) 22 23 24 class TestBuildGoogleClientConfig: 25 def test_structure_matches_installed_app_spec(self) -> None: 26 config = build_google_client_config( 27 client_id="cid.apps.googleusercontent.com", 28 client_secret="SECRET", 29 ) 30 assert "installed" in config 31 inst = config["installed"] 32 assert inst["client_id"] == "cid.apps.googleusercontent.com" 33 assert inst["client_secret"] == "SECRET" 34 assert inst["auth_uri"] == "https://accounts.google.com/o/oauth2/auth" 35 assert inst["token_uri"] == "https://oauth2.googleapis.com/token" 36 # Redirect URIs must include localhost so the interactive 37 # InstalledAppFlow path still works. 38 assert "http://localhost" in inst["redirect_uris"] 39 40 41 class TestValidateLocalRedirectUri: 42 """The redirect_uri is enforced as localhost-only so a prompt- 43 injected agent cannot redirect the OAuth grant to a remote host.""" 44 45 @pytest.mark.parametrize( 46 "bad", 47 [ 48 "https://127.0.0.1:8080/cb", # wrong scheme 49 "http://evil.example.com/cb", # remote host 50 "http://attacker.localhost.com/", # suffix trick 51 "javascript:alert(1)", 52 "file:///etc/passwd", 53 "ftp://localhost/cb", 54 ], 55 ) 56 def test_rejects_non_localhost(self, bad: str) -> None: 57 with pytest.raises(ValueError): 58 build_google_flow( 59 client_id="cid", client_secret="SECRET", redirect_uri=bad 60 ) 61 62 @pytest.mark.parametrize( 63 "good", 64 [ 65 "http://127.0.0.1:8080/cb", 66 "http://localhost:59999/google-ads/callback", 67 "http://127.0.0.1:0/", 68 ], 69 ) 70 def test_accepts_localhost(self, good: str) -> None: 71 flow = build_google_flow( 72 client_id="cid", client_secret="SECRET", redirect_uri=good 73 ) 74 assert flow.redirect_uri == good 75 76 77 class TestBuildGoogleClientConfigImmutable: 78 def test_successive_calls_return_independent_dicts(self) -> None: 79 a = build_google_client_config("cid", "S1") 80 b = build_google_client_config("cid", "S2") 81 # Mutating one must not leak into the other. 82 a["installed"]["redirect_uris"].append("http://evil.example.com") 83 assert "http://evil.example.com" not in b["installed"]["redirect_uris"] 84 85 86 class TestBuildGoogleFlow: 87 def test_no_redirect_uri_returns_installed_app_flow(self) -> None: 88 """When redirect_uri is None, the CLI path uses InstalledAppFlow 89 (which spins up its own local server).""" 90 from google_auth_oauthlib.flow import InstalledAppFlow 91 92 flow = build_google_flow( 93 client_id="cid", client_secret="SECRET", redirect_uri=None 94 ) 95 assert isinstance(flow, InstalledAppFlow) 96 97 def test_with_redirect_uri_returns_plain_flow(self) -> None: 98 """With a redirect_uri, the web wizard path uses plain Flow so 99 the caller handles the callback on its own HTTP server.""" 100 from google_auth_oauthlib.flow import Flow, InstalledAppFlow 101 102 flow = build_google_flow( 103 client_id="cid", 104 client_secret="SECRET", 105 redirect_uri="http://127.0.0.1:59999/google-ads/callback", 106 ) 107 assert isinstance(flow, Flow) 108 assert not isinstance(flow, InstalledAppFlow) 109 assert flow.redirect_uri == ( 110 "http://127.0.0.1:59999/google-ads/callback" 111 ) 112 113 def test_scopes_cover_google_ads_and_search_console(self) -> None: 114 """Scopes must include both Google Ads and Search Console so 115 a single refresh_token drives both MCP tool surfaces.""" 116 flow = build_google_flow( 117 client_id="cid", 118 client_secret="SECRET", 119 redirect_uri="http://127.0.0.1:1/cb", 120 ) 121 url, _state = flow.authorization_url(access_type="offline") 122 assert "adwords" in url 123 assert "webmasters" in url 124 125 126 class TestGoogleAuthUrl: 127 def test_returns_url_and_state(self) -> None: 128 flow = build_google_flow( 129 client_id="cid", 130 client_secret="SECRET", 131 redirect_uri="http://127.0.0.1:1/cb", 132 ) 133 url, state = google_auth_url(flow) 134 assert url.startswith("https://accounts.google.com/") 135 assert isinstance(state, str) and len(state) > 0 136 137 def test_forces_offline_and_consent(self) -> None: 138 """``access_type=offline`` + ``prompt=consent`` guarantees that 139 Google returns a refresh_token every time, even if the user 140 already granted the app before.""" 141 flow = build_google_flow( 142 client_id="cid", 143 client_secret="SECRET", 144 redirect_uri="http://127.0.0.1:1/cb", 145 ) 146 url, _ = google_auth_url(flow) 147 assert "access_type=offline" in url 148 assert "prompt=consent" in url 149 150 151 class TestExchangeGoogleCode: 152 def test_returns_oauth_result_with_refresh_token(self) -> None: 153 """Happy path: Flow's ``fetch_token`` populates credentials, 154 and ``exchange_google_code`` returns an ``OAuthResult``.""" 155 from mureo.auth_setup import OAuthResult 156 157 flow = MagicMock() 158 flow.credentials.refresh_token = "REFRESH_123" 159 flow.credentials.token = "ACCESS_ABC" 160 161 result = exchange_google_code(flow, code="AUTH_CODE") 162 163 flow.fetch_token.assert_called_once_with(code="AUTH_CODE") 164 assert isinstance(result, OAuthResult) 165 assert result.refresh_token == "REFRESH_123" 166 assert result.access_token == "ACCESS_ABC" 167 168 def test_missing_refresh_token_raises(self) -> None: 169 """Google only returns a refresh_token when ``prompt=consent`` 170 is forced. If it comes back missing the exchange must fail 171 loudly so the caller can re-run with the right prompt.""" 172 flow = MagicMock() 173 flow.credentials.refresh_token = None 174 flow.credentials.token = "ACCESS_ABC" 175 176 with pytest.raises(RuntimeError, match="refresh_token"): 177 exchange_google_code(flow, code="AUTH_CODE") 178 179 def test_missing_refresh_token_message_is_actionable(self) -> None: 180 """Operator needs to know HOW to fix it. Check the message 181 mentions prompt=consent (so they know where to look).""" 182 flow = MagicMock() 183 flow.credentials.refresh_token = None 184 flow.credentials.token = "ACCESS_ABC" 185 186 with pytest.raises(RuntimeError) as exc_info: 187 exchange_google_code(flow, code="AUTH_CODE") 188 msg = str(exc_info.value) 189 assert "prompt=consent" in msg 190 assert "access_type=offline" in msg 191 192 def test_fetch_token_error_propagates(self) -> None: 193 """Network / invalid-code errors from fetch_token must not be 194 swallowed — they are distinct from the 'no refresh_token' 195 case and need the caller's own retry / reporting logic.""" 196 197 class _NetworkError(Exception): 198 pass 199 200 flow = MagicMock() 201 flow.fetch_token.side_effect = _NetworkError("connection reset") 202 203 with pytest.raises(_NetworkError, match="connection reset"): 204 exchange_google_code(flow, code="AUTH_CODE") 205 206 207 class TestRunGoogleOauthUsesNewHelpers: 208 """Regression lock: after refactor, ``run_google_oauth`` must still 209 work (it is monkeypatched by many existing setup tests). The new 210 implementation delegates to ``build_google_flow`` internally but 211 keeps the same public behavior. 212 """ 213 214 @pytest.mark.asyncio 215 async def test_returns_oauth_result(self) -> None: 216 from mureo.auth_setup import OAuthResult, run_google_oauth 217 218 mock_creds = MagicMock() 219 mock_creds.refresh_token = "REFRESH" 220 mock_creds.token = "ACCESS" 221 222 with patch( 223 "mureo.auth_setup.build_google_flow" 224 ) as mock_build_flow: 225 mock_flow = MagicMock() 226 mock_flow.run_local_server.return_value = mock_creds 227 mock_build_flow.return_value = mock_flow 228 229 result = await run_google_oauth( 230 client_id="cid", client_secret="SECRET" 231 ) 232 233 assert isinstance(result, OAuthResult) 234 assert result.refresh_token == "REFRESH" 235 assert result.access_token == "ACCESS" 236 # Calls build_google_flow with redirect_uri=None (interactive path). 237 mock_build_flow.assert_called_once_with( 238 client_id="cid", client_secret="SECRET", redirect_uri=None 239 )