room_cal/tests/test_refresh_cookies.py

import asyncio
import os
import re
import shutil
import subprocess
import tempfile
import warnings
from datetime import date, timedelta
from importlib.resources import files
from pathlib import Path

import pytest

from room_cal._utils import data_home as _data_home, get_config
from room_cal.service_api.types import ServiceAuthError


def _ts_dir() -> Path:
  return Path(str(files('room_cal.service_api').joinpath('typescript')))


def _browsers_path() -> Path:
  return _data_home() / '.playwright-browsers'


def _npm_env() -> dict:
  env = os.environ.copy()
  env.setdefault('ROOM_CAL_DATA_HOME', str(_data_home()))
  env['PLAYWRIGHT_BROWSERS_PATH'] = str(_browsers_path())
  return env


def _npm_run(script: str, timeout: int, extra_env: dict | None = None) -> subprocess.CompletedProcess:
  ts_dir = _ts_dir()
  if not (ts_dir / 'node_modules').exists():
    subprocess.run(['npm', 'install'], cwd=ts_dir, check=True, capture_output=True)
  env = _npm_env()
  if extra_env:
    env.update(extra_env)
  try:
    return subprocess.run(
      ['npm', 'run', script],
      cwd=ts_dir,
      env=env,
      capture_output=True,
      text=True,
      timeout=timeout,
    )
  except subprocess.TimeoutExpired as exc:
    pytest.fail(f'npm run {script} timed out after {exc.timeout}s')
  except Exception as exc:
    pytest.fail(f'Failed to launch npm run {script}: {exc}')


@pytest.fixture(scope='session')
def chrome_installed():
  result = _npm_run('install-browsers', timeout=30)
  if result.returncode != 0:
    pytest.fail(
      f'npm run install-browsers failed (exit {result.returncode})\n'
      f'stdout:\n{result.stdout}\nstderr:\n{result.stderr}'
    )
  browsers = _browsers_path()
  chromium_dirs = [p for p in browsers.iterdir() if p.is_dir() and 'chromium' in p.name] if browsers.exists() else []
  if not chromium_dirs:
    pytest.fail(f'install-browsers exited 0 but no chromium dirs found in {browsers}')


def test_refresh_all_flags_match_config():
  """ENABLED flags in refresh-all.ts must match disabled in config.json, with full coverage."""
  ts_path = _ts_dir() / 'scripts' / 'refresh-all.ts'
  src = ts_path.read_text()

  # Parse: ENABLED: Record<string, boolean> = { key: true/false, ... }
  block = re.search(r'const ENABLED[^=]*=\s*\{([^}]+)\}', src)
  assert block, 'Could not find ENABLED block in refresh-all.ts'
  ts_flags: dict[str, bool] = {}
  for m in re.finditer(r'(\w+)\s*:\s*(true|false)', block.group(1)):
    ts_flags[m.group(1)] = m.group(2) == 'true'

  cfg_services = get_config().services
  cfg_keys = set(cfg_services.keys())
  ts_keys = set(ts_flags.keys())

  missing = cfg_keys - ts_keys
  assert not missing, f'Services in config.json missing from refresh-all.ts ENABLED: {missing}'

  extra = ts_keys - cfg_keys
  assert not extra, f'Services in refresh-all.ts ENABLED not in config.json: {extra}'

  mismatches = {
    k for k in cfg_keys
    if ts_flags[k] != (not cfg_services[k].disabled)
  }
  assert not mismatches, (
    f'ENABLED flags in refresh-all.ts do not match config.json disabled for: {mismatches}\n'
    + '\n'.join(f'  {k}: ts={ts_flags[k]} config_disabled={cfg_services[k].disabled}' for k in sorted(mismatches))
  )


def test_install_chrome(chrome_installed):
  """npm run install-browsers puts Chromium binaries in the expected location."""


def test_refresh_cookies_headless(chrome_installed):
  """Cookie-refresh machinery: Chromium must launch and write cookies.

  Pass         — cookies updated and API call succeeds.
  Pass + warn  — script ran correctly but session is stale (exit 2), cookies
                 unchanged, or API auth failed after writing cookies.
  Fail         — unexpected error or non-zero exit.
  """
  data = _data_home()
  airbnb_env = data / 'airbnb.env'
  booking_env = data / 'booking.env'

  before = {p: p.read_text() if p.exists() else None for p in (airbnb_env, booking_env)}

  result = _npm_run('refresh-all', timeout=60, extra_env={'HEADLESS': '1'})
  output = f'stdout:\n{result.stdout}\nstderr:\n{result.stderr}'

  if result.returncode == 2:
    warnings.warn(
      'Session is stale — run `refresh-creds` to re-authenticate.\n' + output,
      UserWarning,
      stacklevel=1,
    )
    return

  if result.returncode != 0:
    pytest.fail(f'npm run refresh-all exited {result.returncode}\n' + output)

  after = {p: p.read_text() if p.exists() else None for p in (airbnb_env, booking_env)}
  if not any(after[p] != before[p] for p in (airbnb_env, booking_env)):
    warnings.warn(
      'refresh-all exited 0 but neither airbnb.env nor booking.env changed.',
      UserWarning,
      stacklevel=1,
    )
    return

  try:
    from room_cal.service_api import ServiceAPI
    api = ServiceAPI()
    asyncio.run(api.get_prices(date.today() + timedelta(days=1)))
  except ServiceAuthError as exc:
    warnings.warn(
      f'Cookies were updated but API auth failed ({exc}); session may be invalid.',
      UserWarning,
      stacklevel=1,
    )


# ---------------------------------------------------------------------------
# Live auto-login test — opt-in only, not run in normal CI
# Set BOOKING_LIVE_TEST=1 to enable.
# Requires BOOKING_EMAIL and BOOKING_PASSWORD in booking.env.
# ---------------------------------------------------------------------------

def _booking_env_credentials() -> tuple[str, str]:
  """Return (email, password) from booking.env, or ('', '') if absent."""
  env_path = _data_home() / 'booking.env'
  if not env_path.exists():
    return '', ''
  text = env_path.read_text()
  email = next((m.group(1) for m in [re.search(r'^BOOKING_EMAIL=(\S+)', text, re.M)] if m), '')
  password = next((m.group(1) for m in [re.search(r'^BOOKING_PASSWORD=(\S+)', text, re.M)] if m), '')
  return email, password


@pytest.mark.skipif(
  not os.environ.get('BOOKING_LIVE_TEST'),
  reason='Set BOOKING_LIVE_TEST=1 to run live Booking.com auto-login test',
)
def test_booking_auto_login(chrome_installed):
  """Full auto-login flow from an empty session.

  Wipes the Booking.com Playwright profile and blanks the session credentials
  in a temporary DATA_HOME, then runs refresh-booking. Asserts that fresh
  cookies and a session token are written, and that the API can make a live
  calendar call with them.
  """
  email, password = _booking_env_credentials()
  if not email or not password:
    pytest.skip('BOOKING_EMAIL and BOOKING_PASSWORD must be set in booking.env')

  real_data = _data_home()
  booking_env_src = real_data / 'booking.env'

  with tempfile.TemporaryDirectory(prefix='room_cal_login_test_') as tmp:
    tmp_path = Path(tmp)

    # Copy booking.env into the temp DATA_HOME, stripping session credentials
    # so the script has no valid session to reuse.
    src_text = booking_env_src.read_text() if booking_env_src.exists() else ''
    stripped = re.sub(r'^(BOOKING_COOKIES|BOOKING_SESSION|BOOKING_CSRF_TOKEN)=.*$', r'\1=', src_text, flags=re.M)
    (tmp_path / 'booking.env').write_text(stripped)

    # Copy browsers so the test doesn't re-download Chromium.
    browsers_src = _browsers_path()
    if browsers_src.exists():
      shutil.copytree(browsers_src, tmp_path / '.playwright-browsers', symlinks=True)

    # Clear the login-attempts counter so rate limiting doesn't interfere.
    attempts_file = tmp_path / 'booking-login-attempts.json'
    if attempts_file.exists():
      attempts_file.unlink()

    result = _npm_run('refresh-booking', timeout=180, extra_env={
      'ROOM_CAL_DATA_HOME': str(tmp_path),
      'PLAYWRIGHT_BROWSERS_PATH': str(tmp_path / '.playwright-browsers'),
    })
    output = f'stdout:\n{result.stdout}\nstderr:\n{result.stderr}'

    assert result.returncode == 0, f'refresh-booking exited {result.returncode}\n{output}'

    new_text = (tmp_path / 'booking.env').read_text()
    new_cookies = re.search(r'^BOOKING_COOKIES=(\S+)', new_text, re.M)
    new_session = re.search(r'^BOOKING_SESSION=(\S+)', new_text, re.M)

    assert new_cookies and new_cookies.group(1), f'No BOOKING_COOKIES written\n{output}'
    assert new_session and new_session.group(1), f'No BOOKING_SESSION written\n{output}'

    # Copy fresh credentials into the real DATA_HOME so the API call below uses them.
    booking_env_src.write_text(new_text)

  from room_cal.service_api import ServiceAPI
  api = ServiceAPI()
  prices = asyncio.run(api.get_prices(date.today() + timedelta(days=1)))
  booking_prices = {d: p for d, p in prices.items() if d}
  assert booking_prices, 'API returned no prices after auto-login — session may not have been accepted'