test_refresh_cookies.py 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203
  1. import asyncio
  2. import os
  3. import re
  4. import shutil
  5. import subprocess
  6. import tempfile
  7. import warnings
  8. from datetime import date, timedelta
  9. from importlib.resources import files
  10. from pathlib import Path
  11. import pytest
  12. from room_cal._utils import data_home as _data_home, get_config
  13. from room_cal.service_api.types import ServiceAuthError
  14. pytestmark = pytest.mark.live
  15. def _ts_dir() -> Path:
  16. return Path(str(files('room_cal.service_api').joinpath('typescript')))
  17. def _browsers_path() -> Path:
  18. return _data_home() / '.playwright-browsers'
  19. def _npm_env() -> dict:
  20. env = os.environ.copy()
  21. env.setdefault('ROOM_CAL_DATA_HOME', str(_data_home()))
  22. env['PLAYWRIGHT_BROWSERS_PATH'] = str(_browsers_path())
  23. return env
  24. def _npm_run(script: str, timeout: int, extra_env: dict | None = None) -> subprocess.CompletedProcess:
  25. ts_dir = _ts_dir()
  26. if not (ts_dir / 'node_modules').exists():
  27. subprocess.run(['npm', 'install'], cwd=ts_dir, check=True, capture_output=True)
  28. env = _npm_env()
  29. if extra_env:
  30. env.update(extra_env)
  31. try:
  32. return subprocess.run(
  33. ['npm', 'run', script],
  34. cwd=ts_dir,
  35. env=env,
  36. capture_output=True,
  37. text=True,
  38. timeout=timeout,
  39. )
  40. except subprocess.TimeoutExpired as exc:
  41. pytest.fail(f'npm run {script} timed out after {exc.timeout}s')
  42. except Exception as exc:
  43. pytest.fail(f'Failed to launch npm run {script}: {exc}')
  44. @pytest.fixture(scope='session')
  45. def chrome_installed():
  46. result = _npm_run('install-browsers', timeout=30)
  47. if result.returncode != 0:
  48. pytest.fail(
  49. f'npm run install-browsers failed (exit {result.returncode})\n'
  50. f'stdout:\n{result.stdout}\nstderr:\n{result.stderr}'
  51. )
  52. browsers = _browsers_path()
  53. chromium_dirs = [p for p in browsers.iterdir() if p.is_dir() and 'chromium' in p.name] if browsers.exists() else []
  54. if not chromium_dirs:
  55. pytest.fail(f'install-browsers exited 0 but no chromium dirs found in {browsers}')
  56. def test_install_chrome(chrome_installed):
  57. """npm run install-browsers puts Chromium binaries in the expected location."""
  58. def test_refresh_cookies_headless(chrome_installed):
  59. """Cookie-refresh machinery: Chromium must launch and write cookies.
  60. Pass — cookies updated and API call succeeds.
  61. Pass + warn — script ran correctly but session is stale (exit 2), cookies
  62. unchanged, or API auth failed after writing cookies.
  63. Fail — unexpected error or non-zero exit.
  64. """
  65. data = _data_home()
  66. airbnb_env = data / 'airbnb.env'
  67. booking_env = data / 'booking.env'
  68. before = {p: p.read_text() if p.exists() else None for p in (airbnb_env, booking_env)}
  69. result = _npm_run('refresh-all', timeout=60, extra_env={'HEADLESS': '1'})
  70. output = f'stdout:\n{result.stdout}\nstderr:\n{result.stderr}'
  71. if result.returncode == 2:
  72. warnings.warn(
  73. 'Session is stale — run `refresh-creds` to re-authenticate.\n' + output,
  74. UserWarning,
  75. stacklevel=1,
  76. )
  77. return
  78. if result.returncode != 0:
  79. pytest.fail(f'npm run refresh-all exited {result.returncode}\n' + output)
  80. after = {p: p.read_text() if p.exists() else None for p in (airbnb_env, booking_env)}
  81. if not any(after[p] != before[p] for p in (airbnb_env, booking_env)):
  82. warnings.warn(
  83. 'refresh-all exited 0 but neither airbnb.env nor booking.env changed.',
  84. UserWarning,
  85. stacklevel=1,
  86. )
  87. return
  88. try:
  89. from room_cal.service_api import ServiceAPI
  90. api = ServiceAPI()
  91. asyncio.run(api.get_prices(date.today() + timedelta(days=1)))
  92. except ServiceAuthError as exc:
  93. warnings.warn(
  94. f'Cookies were updated but API auth failed ({exc}); session may be invalid.',
  95. UserWarning,
  96. stacklevel=1,
  97. )
  98. # ---------------------------------------------------------------------------
  99. # Live auto-login test — opt-in only, not run in normal CI
  100. # Set BOOKING_LIVE_TEST=1 to enable.
  101. # Requires BOOKING_EMAIL and BOOKING_PASSWORD in booking.env.
  102. # ---------------------------------------------------------------------------
  103. def _booking_env_credentials() -> tuple[str, str]:
  104. """Return (email, password) from booking.env, or ('', '') if absent."""
  105. env_path = _data_home() / 'booking.env'
  106. if not env_path.exists():
  107. return '', ''
  108. text = env_path.read_text()
  109. email = next((m.group(1) for m in [re.search(r'^BOOKING_EMAIL=(\S+)', text, re.M)] if m), '')
  110. password = next((m.group(1) for m in [re.search(r'^BOOKING_PASSWORD=(\S+)', text, re.M)] if m), '')
  111. return email, password
  112. @pytest.mark.skipif(
  113. not os.environ.get('BOOKING_LIVE_TEST'),
  114. reason='Set BOOKING_LIVE_TEST=1 to run live Booking.com auto-login test',
  115. )
  116. def test_booking_auto_login(chrome_installed):
  117. """Full auto-login flow from an empty session.
  118. Wipes the Booking.com Playwright profile and blanks the session credentials
  119. in a temporary DATA_HOME, then runs refresh-booking. Asserts that fresh
  120. cookies and a session token are written, and that the API can make a live
  121. calendar call with them.
  122. """
  123. email, password = _booking_env_credentials()
  124. if not email or not password:
  125. pytest.skip('BOOKING_EMAIL and BOOKING_PASSWORD must be set in booking.env')
  126. real_data = _data_home()
  127. booking_env_src = real_data / 'booking.env'
  128. with tempfile.TemporaryDirectory(prefix='room_cal_login_test_') as tmp:
  129. tmp_path = Path(tmp)
  130. # Copy booking.env into the temp DATA_HOME, stripping session credentials
  131. # so the script has no valid session to reuse.
  132. src_text = booking_env_src.read_text() if booking_env_src.exists() else ''
  133. stripped = re.sub(r'^(BOOKING_COOKIES|BOOKING_SESSION|BOOKING_CSRF_TOKEN)=.*$', r'\1=', src_text, flags=re.M)
  134. (tmp_path / 'booking.env').write_text(stripped)
  135. # Copy browsers so the test doesn't re-download Chromium.
  136. browsers_src = _browsers_path()
  137. if browsers_src.exists():
  138. shutil.copytree(browsers_src, tmp_path / '.playwright-browsers', symlinks=True)
  139. # Clear the login-attempts counter so rate limiting doesn't interfere.
  140. attempts_file = tmp_path / 'booking-login-attempts.json'
  141. if attempts_file.exists():
  142. attempts_file.unlink()
  143. result = _npm_run('refresh-booking', timeout=180, extra_env={
  144. 'ROOM_CAL_DATA_HOME': str(tmp_path),
  145. 'PLAYWRIGHT_BROWSERS_PATH': str(tmp_path / '.playwright-browsers'),
  146. })
  147. output = f'stdout:\n{result.stdout}\nstderr:\n{result.stderr}'
  148. assert result.returncode == 0, f'refresh-booking exited {result.returncode}\n{output}'
  149. new_text = (tmp_path / 'booking.env').read_text()
  150. new_cookies = re.search(r'^BOOKING_COOKIES=(\S+)', new_text, re.M)
  151. new_session = re.search(r'^BOOKING_SESSION=(\S+)', new_text, re.M)
  152. assert new_cookies and new_cookies.group(1), f'No BOOKING_COOKIES written\n{output}'
  153. assert new_session and new_session.group(1), f'No BOOKING_SESSION written\n{output}'
  154. # Copy fresh credentials into the real DATA_HOME so the API call below uses them.
  155. booking_env_src.write_text(new_text)
  156. from room_cal.service_api import ServiceAPI
  157. api = ServiceAPI()
  158. prices = asyncio.run(api.get_prices(date.today() + timedelta(days=1)))
  159. booking_prices = {d: p for d, p in prices.items() if d}
  160. assert booking_prices, 'API returned no prices after auto-login — session may not have been accepted'