test_meta_token_refresh.py
1 """Tests for Meta Ads Long-Lived Token auto-refresh (TDD: RED phase)""" 2 3 from __future__ import annotations 4 5 import json 6 from datetime import datetime, timedelta, timezone 7 from typing import TYPE_CHECKING, Any 8 from unittest.mock import AsyncMock, patch 9 10 if TYPE_CHECKING: 11 from pathlib import Path 12 13 import httpx 14 import pytest 15 16 from mureo.auth import MetaAdsCredentials, refresh_meta_token_if_needed 17 18 19 @pytest.fixture(autouse=True) 20 def _mock_save_token(request): 21 """Prevent tests from writing to real ~/.mureo/credentials.json. 22 23 Tests that explicitly use tmp_path for credential file operations 24 can opt out by using @pytest.mark.real_save marker. 25 """ 26 if "real_save" in {m.name for m in request.node.iter_markers()}: 27 yield 28 else: 29 with patch("mureo.auth._save_meta_token"): 30 yield 31 32 33 def _now_iso() -> str: 34 return datetime.now(tz=timezone.utc).isoformat() 35 36 37 def _days_ago_iso(days: int) -> str: 38 return (datetime.now(tz=timezone.utc) - timedelta(days=days)).isoformat() 39 40 41 def _make_creds( 42 *, 43 access_token: str = "old-token", 44 app_id: str | None = "app-123", 45 app_secret: str | None = "secret-456", 46 token_obtained_at: str | None = None, 47 ) -> MetaAdsCredentials: 48 return MetaAdsCredentials( 49 access_token=access_token, 50 app_id=app_id, 51 app_secret=app_secret, 52 token_obtained_at=token_obtained_at, 53 ) 54 55 56 def _write_credentials(path: Path, meta_section: dict[str, Any]) -> None: 57 data = {"meta_ads": meta_section} 58 path.parent.mkdir(parents=True, exist_ok=True) 59 path.write_text(json.dumps(data), encoding="utf-8") 60 61 62 # --------------------------------------------------------------------------- 63 # 1. No refresh when token is fresh (10 days old) 64 # --------------------------------------------------------------------------- 65 66 67 @pytest.mark.unit 68 async def test_no_refresh_when_token_is_fresh() -> None: 69 """Token obtained 10 days ago should NOT be refreshed.""" 70 creds = _make_creds(token_obtained_at=_days_ago_iso(10)) 71 72 result = await refresh_meta_token_if_needed(creds) 73 74 assert result is creds # Same object, no refresh occurred 75 76 77 # --------------------------------------------------------------------------- 78 # 2. Refresh when token is expiring soon (55 days old) 79 # --------------------------------------------------------------------------- 80 81 82 @pytest.mark.unit 83 async def test_refresh_when_token_expiring_soon() -> None: 84 """Token obtained 55 days ago (>53 threshold) should trigger refresh.""" 85 creds = _make_creds(token_obtained_at=_days_ago_iso(55)) 86 87 mock_response = httpx.Response( 88 200, 89 json={ 90 "access_token": "new-refreshed-token", 91 "token_type": "bearer", 92 "expires_in": 5183944, 93 }, 94 request=httpx.Request("GET", "https://example.com"), 95 ) 96 97 with patch("mureo.auth.httpx.AsyncClient") as mock_client_cls: 98 mock_client = AsyncMock() 99 mock_client.get.return_value = mock_response 100 mock_client.__aenter__ = AsyncMock(return_value=mock_client) 101 mock_client.__aexit__ = AsyncMock(return_value=False) 102 mock_client_cls.return_value = mock_client 103 104 result = await refresh_meta_token_if_needed(creds) 105 106 assert result.access_token == "new-refreshed-token" 107 assert result.token_obtained_at is not None 108 assert result.token_obtained_at != creds.token_obtained_at 109 110 111 # --------------------------------------------------------------------------- 112 # 3. No refresh without app credentials 113 # --------------------------------------------------------------------------- 114 115 116 @pytest.mark.unit 117 async def test_no_refresh_without_app_id() -> None: 118 """If app_id is None, skip refresh.""" 119 creds = _make_creds( 120 app_id=None, 121 token_obtained_at=_days_ago_iso(55), 122 ) 123 124 result = await refresh_meta_token_if_needed(creds) 125 126 assert result is creds 127 128 129 @pytest.mark.unit 130 async def test_no_refresh_without_app_secret() -> None: 131 """If app_secret is None, skip refresh.""" 132 creds = _make_creds( 133 app_secret=None, 134 token_obtained_at=_days_ago_iso(55), 135 ) 136 137 result = await refresh_meta_token_if_needed(creds) 138 139 assert result is creds 140 141 142 # --------------------------------------------------------------------------- 143 # 4. No refresh without token_obtained_at 144 # --------------------------------------------------------------------------- 145 146 147 @pytest.mark.unit 148 async def test_no_refresh_without_obtained_at() -> None: 149 """If token_obtained_at is None, skip refresh.""" 150 creds = _make_creds(token_obtained_at=None) 151 152 result = await refresh_meta_token_if_needed(creds) 153 154 assert result is creds 155 156 157 # --------------------------------------------------------------------------- 158 # 5. Refresh updates credentials file 159 # --------------------------------------------------------------------------- 160 161 162 @pytest.mark.unit 163 @pytest.mark.real_save 164 async def test_refresh_updates_credentials_file(tmp_path: Path) -> None: 165 """After refresh, credentials.json should contain the new token.""" 166 cred_path = tmp_path / "credentials.json" 167 _write_credentials( 168 cred_path, 169 { 170 "access_token": "old-token", 171 "app_id": "app-123", 172 "app_secret": "secret-456", 173 "token_obtained_at": _days_ago_iso(55), 174 }, 175 ) 176 177 creds = _make_creds(token_obtained_at=_days_ago_iso(55)) 178 179 mock_response = httpx.Response( 180 200, 181 json={ 182 "access_token": "new-refreshed-token", 183 "token_type": "bearer", 184 "expires_in": 5183944, 185 }, 186 request=httpx.Request("GET", "https://example.com"), 187 ) 188 189 with patch("mureo.auth.httpx.AsyncClient") as mock_client_cls: 190 mock_client = AsyncMock() 191 mock_client.get.return_value = mock_response 192 mock_client.__aenter__ = AsyncMock(return_value=mock_client) 193 mock_client.__aexit__ = AsyncMock(return_value=False) 194 mock_client_cls.return_value = mock_client 195 196 await refresh_meta_token_if_needed(creds, path=cred_path) 197 198 # Verify file was updated 199 saved_data = json.loads(cred_path.read_text(encoding="utf-8")) 200 assert saved_data["meta_ads"]["access_token"] == "new-refreshed-token" 201 assert "token_obtained_at" in saved_data["meta_ads"] 202 203 204 # --------------------------------------------------------------------------- 205 # 6. Refresh failure returns original credentials 206 # --------------------------------------------------------------------------- 207 208 209 @pytest.mark.unit 210 async def test_refresh_failure_returns_original() -> None: 211 """If API call fails, return original credentials without crashing.""" 212 creds = _make_creds(token_obtained_at=_days_ago_iso(55)) 213 214 with patch("mureo.auth.httpx.AsyncClient") as mock_client_cls: 215 mock_client = AsyncMock() 216 mock_client.get.side_effect = httpx.HTTPError("Network error") 217 mock_client.__aenter__ = AsyncMock(return_value=mock_client) 218 mock_client.__aexit__ = AsyncMock(return_value=False) 219 mock_client_cls.return_value = mock_client 220 221 result = await refresh_meta_token_if_needed(creds) 222 223 assert result is creds 224 225 226 @pytest.mark.unit 227 async def test_refresh_failure_on_non_200_returns_original() -> None: 228 """If API returns non-200, return original credentials.""" 229 creds = _make_creds(token_obtained_at=_days_ago_iso(55)) 230 231 mock_response = httpx.Response( 232 400, 233 json={"error": {"message": "Invalid token"}}, 234 request=httpx.Request("GET", "https://example.com"), 235 ) 236 237 with patch("mureo.auth.httpx.AsyncClient") as mock_client_cls: 238 mock_client = AsyncMock() 239 mock_client.get.return_value = mock_response 240 mock_client.__aenter__ = AsyncMock(return_value=mock_client) 241 mock_client.__aexit__ = AsyncMock(return_value=False) 242 mock_client_cls.return_value = mock_client 243 244 result = await refresh_meta_token_if_needed(creds) 245 246 assert result is creds 247 248 249 # --------------------------------------------------------------------------- 250 # 7. Verify correct API call parameters 251 # --------------------------------------------------------------------------- 252 253 254 @pytest.mark.unit 255 async def test_refresh_api_call_parameters() -> None: 256 """Verify the correct endpoint and params are used for the refresh call.""" 257 creds = _make_creds(token_obtained_at=_days_ago_iso(55)) 258 259 mock_response = httpx.Response( 260 200, 261 json={ 262 "access_token": "new-token", 263 "token_type": "bearer", 264 "expires_in": 5183944, 265 }, 266 request=httpx.Request("GET", "https://example.com"), 267 ) 268 269 with patch("mureo.auth.httpx.AsyncClient") as mock_client_cls: 270 mock_client = AsyncMock() 271 mock_client.get.return_value = mock_response 272 mock_client.__aenter__ = AsyncMock(return_value=mock_client) 273 mock_client.__aexit__ = AsyncMock(return_value=False) 274 mock_client_cls.return_value = mock_client 275 276 await refresh_meta_token_if_needed(creds) 277 278 mock_client.get.assert_called_once_with( 279 "https://graph.facebook.com/v21.0/oauth/access_token", 280 params={ 281 "grant_type": "fb_exchange_token", 282 "client_id": "app-123", 283 "client_secret": "secret-456", 284 "fb_exchange_token": "old-token", 285 }, 286 ) 287 288 289 # --------------------------------------------------------------------------- 290 # 8. Token at exact threshold boundary (53 days) 291 # --------------------------------------------------------------------------- 292 293 294 @pytest.mark.unit 295 async def test_refresh_at_exact_threshold() -> None: 296 """Token exactly 53 days old should trigger refresh.""" 297 creds = _make_creds(token_obtained_at=_days_ago_iso(53)) 298 299 mock_response = httpx.Response( 300 200, 301 json={ 302 "access_token": "refreshed-at-boundary", 303 "token_type": "bearer", 304 "expires_in": 5183944, 305 }, 306 request=httpx.Request("GET", "https://example.com"), 307 ) 308 309 with patch("mureo.auth.httpx.AsyncClient") as mock_client_cls: 310 mock_client = AsyncMock() 311 mock_client.get.return_value = mock_response 312 mock_client.__aenter__ = AsyncMock(return_value=mock_client) 313 mock_client.__aexit__ = AsyncMock(return_value=False) 314 mock_client_cls.return_value = mock_client 315 316 result = await refresh_meta_token_if_needed(creds) 317 318 assert result.access_token == "refreshed-at-boundary" 319 320 321 @pytest.mark.unit 322 async def test_no_refresh_at_52_days() -> None: 323 """Token 52 days old should NOT trigger refresh (below 53-day threshold).""" 324 creds = _make_creds(token_obtained_at=_days_ago_iso(52)) 325 326 result = await refresh_meta_token_if_needed(creds) 327 328 assert result is creds 329 330 331 # --------------------------------------------------------------------------- 332 # 9. Credentials file preserves other sections 333 # --------------------------------------------------------------------------- 334 335 336 @pytest.mark.unit 337 @pytest.mark.real_save 338 async def test_refresh_preserves_other_credential_sections( 339 tmp_path: Path, 340 ) -> None: 341 """Refreshing Meta token should not clobber google_ads section.""" 342 cred_path = tmp_path / "credentials.json" 343 full_data = { 344 "google_ads": {"developer_token": "keep-me"}, 345 "meta_ads": { 346 "access_token": "old-token", 347 "app_id": "app-123", 348 "app_secret": "secret-456", 349 "token_obtained_at": _days_ago_iso(55), 350 }, 351 } 352 cred_path.write_text(json.dumps(full_data), encoding="utf-8") 353 354 creds = _make_creds(token_obtained_at=_days_ago_iso(55)) 355 356 mock_response = httpx.Response( 357 200, 358 json={ 359 "access_token": "new-token", 360 "token_type": "bearer", 361 "expires_in": 5183944, 362 }, 363 request=httpx.Request("GET", "https://example.com"), 364 ) 365 366 with patch("mureo.auth.httpx.AsyncClient") as mock_client_cls: 367 mock_client = AsyncMock() 368 mock_client.get.return_value = mock_response 369 mock_client.__aenter__ = AsyncMock(return_value=mock_client) 370 mock_client.__aexit__ = AsyncMock(return_value=False) 371 mock_client_cls.return_value = mock_client 372 373 await refresh_meta_token_if_needed(creds, path=cred_path) 374 375 saved_data = json.loads(cred_path.read_text(encoding="utf-8")) 376 assert saved_data["google_ads"]["developer_token"] == "keep-me" 377 assert saved_data["meta_ads"]["access_token"] == "new-token"