diff --git a/requirements.txt b/requirements.txt index 6e37003..915d297 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,3 +2,5 @@ playwright>=1.40.0 httpx>=0.25.0 click>=8.1.0 openpyxl>=3.1.0 +respx>=0.20.0 +pytest>=7.0.0 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..296bbbc --- /dev/null +++ b/src/__init__.py @@ -0,0 +1,3 @@ +"""Cantaloupe Downloader - Export automation for mycantaloupe.com.""" + +__version__ = "0.1.0" diff --git a/src/auth.py b/src/auth.py new file mode 100644 index 0000000..9a733cb --- /dev/null +++ b/src/auth.py @@ -0,0 +1,370 @@ +""" +Authentication module for mycantaloupe.com. + +Provides: +- HttpxAuth: Lightweight session-based auth using httpx (no browser needed) +- PlaywrightAuth: Browser-based auth using Playwright (fallback) +- get_session(): Convenience factory returning an authenticated httpx.Client +""" + +import os +import logging +from typing import Optional +from dataclasses import dataclass, field + +import httpx + +logger = logging.getLogger(__name__) + +BASE_URL = "https://mycantaloupe.com" +LOGIN_URL = f"{BASE_URL}/login/" +SIGNIN_API = f"{BASE_URL}/login/api/SignIn/SignIn/en-US" +ANTI_FORGERY_API = f"{BASE_URL}/login/api/DataContext/GenerateAntiForgeryState" + +# Session cookies that indicate an authenticated session +AUTH_COOKIE_NAMES = {"ApplicationGatewayAffinity", "AeonPrincipalCacheVersion"} + + +class AuthenticationError(Exception): + """Raised when login fails.""" + + +@dataclass +class AuthSession: + """Holds an authenticated httpx.Client and metadata.""" + client: httpx.Client + email: str + password: str = "" + cookies: dict = field(default_factory=dict) + + def refresh(self) -> bool: + """Attempt to refresh the session if the current one is expired. + + Returns True if the session is (or remains) valid, False otherwise. + """ + # Check if we still have auth cookies + for name in AUTH_COOKIE_NAMES: + if name in self.client.cookies: + return True + # Try re-login + try: + new_session = login(self.email, self.password) + self.client = new_session.client + self.cookies = new_session.cookies + return True + except AuthenticationError: + return False + + +def _get_credentials() -> tuple[str, str]: + """Read credentials from environment variables.""" + email = os.environ.get("CANTALOUPE_EMAIL") + password = os.environ.get("CANTALOUPE_PASSWORD") + if not email or not password: + raise AuthenticationError( + "Credentials not found. Set CANTALOUPE_EMAIL and CANTALOUPE_PASSWORD " + "environment variables." + ) + return email, password + + +def _get_anti_forgery_state(client: httpx.Client) -> str: + """Fetch the anti-forgery state token required for login. + + The login API returns a JSON object with the token: + {"HeaderName": "__RequestVerificationToken", "Token": "..."} + """ + resp = client.get(ANTI_FORGERY_API) + resp.raise_for_status() + data = resp.json() + token = data.get("Token", "") + if not token: + raise AuthenticationError( + f"Failed to obtain anti-forgery state token. Response: {resp.text[:200]}" + ) + logger.debug("Obtained anti-forgery state token: %s...", token[:20]) + return token + + +def _do_login( + client: httpx.Client, email: str, password: str, anti_forgery_token: str +) -> dict: + """Perform the actual login POST request. + + Sends credentials + anti-forgery token header. Returns the response + JSON (on success) or raises AuthenticationError. + + The login API responds with one of: + {"SuccessfulResult": {...}} -> success + {"ErrorMessage": "..."} -> failure + {"NotVerifiedUserEmail": "..."} -> email not yet verified + """ + payload = { + "Email": email, + "Password": password, + "FormsAuthReturnUrl": "/cs4/VueMachineList", + } + resp = client.post( + SIGNIN_API, + json=payload, + headers={ + "__RequestVerificationToken": anti_forgery_token, + "Content-Type": "application/json", + "Accept": "application/json, text/plain, */*", + "X-Requested-With": "XMLHttpRequest", + }, + ) + + # The API returns 200 on both success and failure, but the JSON body + # contains an error message on failure. + if resp.status_code != 200: + raise AuthenticationError( + f"Login request failed with status {resp.status_code}: {resp.text[:200]}" + ) + + data = resp.json() + + # Check for error in response + error = data.get("ErrorMessage") + if error: + raise AuthenticationError(f"Login rejected: {error}") + + # Check for not-verified state + not_verified = data.get("NotVerifiedUserEmail") + if not_verified: + raise AuthenticationError( + f"Email not verified: {not_verified}. " + "Check your inbox for a verification email from mycantaloupe.com." + ) + + # Check for success indicator + if data.get("SuccessfulResult") is not None: + logger.info("Login successful for %s", email) + return data + + # Some responses may have a redirect URL variant + if data.get("RedirectUrl") or data.get("redirectUrl"): + logger.info( + "Login successful for %s (redirect: %s)", + email, + data.get("RedirectUrl"), + ) + return data + + # If none of the above, but the response looks valid, assume success + logger.warning( + "Login response unclear (no error, no success flag), but proceeding: %s", + str(data)[:200], + ) + return data + + +def login( + email: Optional[str] = None, + password: Optional[str] = None, + client: Optional[httpx.Client] = None, +) -> AuthSession: + """Authenticate with mycantaloupe.com and return an authenticated session. + + Args: + email: Account email. Falls back to CANTALOUPE_EMAIL env var. + password: Account password. Falls back to CANTALOUPE_PASSWORD env var. + client: Optional pre-configured httpx.Client. If not provided, a new + one is created with default cookie handling. + + Returns: + AuthSession with an authenticated httpx.Client. + + Raises: + AuthenticationError: If credentials are missing or login fails. + """ + if not email or not password: + email, password = _get_credentials() + + if client is None: + client = httpx.Client( + base_url=BASE_URL, + cookies={}, + follow_redirects=True, + ) + + # Step 1: Get anti-forgery token + anti_forgery = _get_anti_forgery_state(client) + + # Step 2: POST login credentials (passes anti-forgery token as header) + _do_login(client, email, password, anti_forgery) + + # Step 3: Collect cookies from the session + cookies = dict(client.cookies) + + return AuthSession(client=client, email=email, password=password, cookies=cookies) + + +def get_session( + email: Optional[str] = None, + password: Optional[str] = None, + force_login: bool = False, +) -> httpx.Client: + """Convenience factory: login and return an authenticated httpx.Client. + + This is the primary entry point for other modules. Call this at startup + to get a ready-to-use authenticated client. + + Args: + email: Account email. Falls back to CANTALOUPE_EMAIL env var. + password: Account password. Falls back to CANTALOUPE_PASSWORD env var. + force_login: If True, always perform a fresh login even if cookies + might already be valid. Default: False. + + Returns: + An authenticated httpx.Client with session cookies. + + Example: + >>> client = get_session() + >>> resp = client.post("/cs4/VueMachineList/ExcelExport") + """ + if not email or not password: + email, password = _get_credentials() + + if force_login: + return login(email, password).client + + # Try with a fresh client first — the server may set auth cookies + # from a prior session if we're on the same IP, but generally we need + # to log in explicitly. + client = httpx.Client( + base_url=BASE_URL, + cookies={}, + follow_redirects=True, + ) + + # Check if we can get to a protected resource without logging in + # (unlikely for this site, but harmless to try before login) + probe = client.get(f"{BASE_URL}/cs4/VueMachineList") + for name in AUTH_COOKIE_NAMES: + if name in client.cookies: + logger.info("Existing session cookies found, reusing session") + return client + + # No valid session, perform login + return login(email, password, client=client).client + + +# --- Playwright-based auth (fallback for JS-heavy flows) --- + +def playwright_login( + email: Optional[str] = None, + password: Optional[str] = None, + headless: bool = True, + storage_state_path: Optional[str] = None, +): + """Authenticate using Playwright browser automation. + + Use this when the httpx-based login fails due to JS challenges, + CAPTCHA, or complex client-side logic. + + Args: + email: Account email. Falls back to CANTALOUPE_EMAIL env var. + password: Account password. Falls back to CANTALOUPE_PASSWORD env var. + headless: Run browser headless (default: True). + storage_state_path: Path to save/load Playwright storage state + (cookies + localStorage). Pass a path to persist + the session for reuse across runs, avoiding + repeated logins. + + Returns: + A tuple of (page, context, browser) — the authenticated Playwright + page, its browser context, and the browser instance. Caller must + close the browser when done. + + Raises: + AuthenticationError: If login fails. + ImportError: If playwright is not installed. + """ + try: + from playwright.sync_api import sync_playwright + except ImportError: + raise ImportError( + "playwright is required for playwright_login(). " + "Install it with: pip install playwright && playwright install chromium" + ) + + if not email or not password: + email, password = _get_credentials() + + with sync_playwright() as p: + browser = p.chromium.launch(headless=headless) + + context_kwargs = {} + if storage_state_path and os.path.exists(storage_state_path): + context_kwargs["storage_state"] = storage_state_path + + context = browser.new_context(**context_kwargs) + page = context.new_page() + + # Navigate to login page + page.goto(LOGIN_URL, wait_until="networkidle") + + # Check if already logged in (storage state restored) + if not page.locator('input[name="email"]').is_visible(timeout=3000): + logger.info("Already authenticated via storage state") + return page, context, browser + + # Fill in credentials + page.fill('input[name="email"]', email) + page.fill('input[name="password"]', password) + + # Click Sign In and wait for navigation + page.click('button:has-text("Sign In")') + + # Wait for either successful redirect or error + try: + page.wait_for_url( + lambda url: "/login/" not in url, + timeout=15000, + ) + logger.info("Playwright login successful for %s", email) + except Exception: + # Check for error message + error_el = page.locator('text=Incorrect username or password') + if error_el.is_visible(timeout=2000): + raise AuthenticationError("Playwright login rejected: incorrect username or password") + raise AuthenticationError("Playwright login failed: timed out waiting for redirect") + + # Save storage state if requested + if storage_state_path: + os.makedirs(os.path.dirname(storage_state_path) or ".", exist_ok=True) + context.storage_state(path=storage_state_path) + logger.info("Saved Playwright storage state to %s", storage_state_path) + + return page, context, browser + + +# --- CLI test helper --- + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + + import sys + + if len(sys.argv) > 1 and sys.argv[1] == "--playwright": + print("Testing Playwright login...") + page, context, browser = playwright_login(headless=False) + print(f"Logged in! Page title: {page.title()}") + browser.close() + else: + print("Testing httpx login...") + try: + session = get_session() + print(f"Logged in! Cookies: {dict(session.cookies)}") + # Probe the machine list page + resp = session.get(f"{BASE_URL}/cs4/VueMachineList") + print(f"Machine list page status: {resp.status_code}") + # Check if we see login redirect + if "login" in resp.url.path.lower(): + print("WARNING: Redirected to login page — session may not be valid") + else: + print(f"Machine list page URL: {resp.url}") + except AuthenticationError as e: + print(f"ERROR: {e}") + sys.exit(1) diff --git a/src/download.py b/src/download.py new file mode 100644 index 0000000..e01af3f --- /dev/null +++ b/src/download.py @@ -0,0 +1,208 @@ +""" +Export/download automation for mycantaloupe.com. + +Authenticates via src.auth and downloads the Machine List Excel export. +Uses httpx directly — no Playwright needed for the HTTP export flow. + +Usage: + python -m src.download # default output dir + python -m src.download --output ~/exports # custom output dir + python -m src.download --output ~/machine-list.xlsx # custom filename +""" + +import os +import logging +import sys +from datetime import datetime +from pathlib import Path +from typing import Optional + +import httpx + +from .auth import get_session + +logger = logging.getLogger(__name__) + +BASE_URL = "https://mycantaloupe.com" +EXPORT_ENDPOINT = "/cs4/VueMachineList/ExcelExport" +MACHINE_LIST_PAGE = "/cs4/VueMachineList" +DEFAULT_OUTPUT_DIR = os.path.expanduser("~/cantaloupe-exports") +DEFAULT_FILENAME = "Machine List.xlsx" + + +class ExportError(Exception): + """Raised when the export download fails.""" + + +def _ensure_authenticated_client() -> httpx.Client: + """Get an authenticated httpx client, attempting login if needed.""" + logger.info("Authenticating with mycantaloupe.com...") + try: + client = get_session() + logger.info("Authentication successful") + return client + except Exception as e: + raise ExportError(f"Authentication failed: {e}") from e + + +def _verify_machine_list_access(client: httpx.Client) -> bool: + """Verify we can access the Machine List page (optional sanity check).""" + resp = client.get(f"{BASE_URL}{MACHINE_LIST_PAGE}", follow_redirects=True) + if resp.status_code >= 400: + logger.warning("Machine List page returned status %d", resp.status_code) + return False + if "/login/" in str(resp.url): + logger.error("Redirected to login — session is not authenticated") + return False + logger.info("Machine List page accessible at %s", resp.url.path) + return True + + +def _download_export(client: httpx.Client) -> bytes: + """POST to the Excel export endpoint and return the response bytes. + + The server generates an .xlsx file server-side and streams it back. + Raises ExportError on failure. + """ + logger.info("Requesting Excel export from %s", EXPORT_ENDPOINT) + + resp = client.post( + f"{BASE_URL}{EXPORT_ENDPOINT}", + follow_redirects=True, + timeout=120, # server-side generation may take time + ) + + if resp.status_code >= 400: + raise ExportError( + f"Export request failed with status {resp.status_code}: " + f"{resp.text[:200]}" + ) + + # Check for redirect to login (session expired) + if "/login/" in str(resp.url) or "/login/" in str(resp.headers.get("location", "")): + raise ExportError("Export redirected to login — session may have expired") + + content_type = resp.headers.get("content-type", "") + logger.info( + "Export response: status=%d, content-type=%s, size=%d bytes", + resp.status_code, + content_type, + len(resp.content), + ) + + if len(resp.content) == 0: + raise ExportError("Export response is empty — server may have rejected the request") + + return resp.content + + +def _resolve_output_path(output: Optional[str] = None) -> Path: + """Resolve the output file path from user input or defaults. + + If output is a directory (or ends with /), the default filename is appended. + If output is a file path, it's used as-is. + Timestamp is added to avoid overwrites. + """ + if output: + out_path = Path(os.path.expanduser(output)) + else: + out_path = Path(DEFAULT_OUTPUT_DIR) + + # If path exists and is a directory, or ends with /, treat as directory + if out_path.is_dir() or str(output or "").endswith(os.sep): + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + name, ext = os.path.splitext(DEFAULT_FILENAME) + out_path = out_path / f"{name}_{timestamp}{ext}" + else: + # Treat as file path — ensure parent dir exists + out_path.parent.mkdir(parents=True, exist_ok=True) + + # Ensure parent directory exists + out_path.parent.mkdir(parents=True, exist_ok=True) + + return out_path + + +def download(output: Optional[str] = None) -> Path: + """Main entry point: authenticate, export, save to disk. + + Args: + output: Output path. Directory or file path. + Defaults to ~/cantaloupe-exports/ with timestamped filename. + + Returns: + Path to the saved .xlsx file. + + Raises: + ExportError: On any failure (auth, export, save). + """ + # 1. Authenticate + client = _ensure_authenticated_client() + + try: + # 2. Optional: verify we can access the machine list + _verify_machine_list_access(client) + + # 3. Download the export + content = _download_export(client) + + # 4. Save to disk + out_path = _resolve_output_path(output) + out_path.write_bytes(content) + logger.info("Saved %d bytes to %s", len(content), out_path) + + finally: + client.close() + + return out_path + + +def download_sync(output: Optional[str] = None) -> Path: + """Synchronous convenience wrapper for download().""" + return download(output=output) + + +# --- CLI --- + +def main(argv: Optional[list] = None) -> int: + """CLI entry point: python -m src.download [--output PATH] [--verbose].""" + import argparse + + parser = argparse.ArgumentParser( + description="Download Machine List Excel export from mycantaloupe.com" + ) + parser.add_argument( + "-o", "--output", + default=None, + help="Output file or directory (default: ~/cantaloupe-exports/)", + ) + parser.add_argument( + "-v", "--verbose", + action="store_true", + help="Enable debug logging", + ) + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + + try: + out_path = download(output=args.output) + print(f"✅ Export saved to: {out_path}") + return 0 + except ExportError as e: + logger.error("Export failed: %s", e) + return 1 + except KeyboardInterrupt: + logger.info("Cancelled by user") + return 130 + except Exception: + logger.exception("Unexpected error during export") + return 2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..30ef467 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Tests for cantaloupe-downloader.""" diff --git a/tests/test_auth.py b/tests/test_auth.py new file mode 100644 index 0000000..b1f98ef --- /dev/null +++ b/tests/test_auth.py @@ -0,0 +1,358 @@ +"""Tests for auth module — unit tests with mocked httpx via respx.""" +import os +from unittest.mock import patch + +import httpx +import pytest +import respx + +from src.auth import ( + ANTI_FORGERY_API, + AUTH_COOKIE_NAMES, + SIGNIN_API, + AuthSession, + AuthenticationError, + _do_login, + _get_anti_forgery_state, + _get_credentials, + get_session, + login, +) + + +# ───────────────────────────────────────────────── +# _get_credentials +# ───────────────────────────────────────────────── + + +def test_get_credentials_from_env(): + with patch.dict(os.environ, {"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"}): + email, password = _get_credentials() + assert email == "u@test.com" + assert password == "pw" + + +def test_get_credentials_missing_email(): + with patch.dict(os.environ, {"CANTALOUPE_PASSWORD": "pw"}, clear=True): + with pytest.raises(AuthenticationError, match="Credentials not found"): + _get_credentials() + + +def test_get_credentials_missing_password(): + with patch.dict(os.environ, {"CANTALOUPE_EMAIL": "u@test.com"}, clear=True): + with pytest.raises(AuthenticationError, match="Credentials not found"): + _get_credentials() + + +def test_get_credentials_none_set(): + with patch.dict(os.environ, {}, clear=True): + with pytest.raises(AuthenticationError, match="Credentials not found"): + _get_credentials() + + +# ───────────────────────────────────────────────── +# _get_anti_forgery_state +# ───────────────────────────────────────────────── + + +@respx.mock +def test_anti_forgery_success(): + token = "CfDJ8ExampleToken12345" + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": token} + ) + client = httpx.Client(base_url="https://mycantaloupe.com") + result = _get_anti_forgery_state(client) + assert result == token + + +@respx.mock +def test_anti_forgery_empty_token(): + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": ""} + ) + client = httpx.Client(base_url="https://mycantaloupe.com") + with pytest.raises(AuthenticationError, match="Failed to obtain anti-forgery"): + _get_anti_forgery_state(client) + + +@respx.mock +def test_anti_forgery_missing_token_key(): + respx.get(ANTI_FORGERY_API).respond(json={"other": "data"}) + client = httpx.Client(base_url="https://mycantaloupe.com") + with pytest.raises(AuthenticationError, match="Failed to obtain anti-forgery"): + _get_anti_forgery_state(client) + + +@respx.mock +def test_anti_forgery_non_200(): + respx.get(ANTI_FORGERY_API).respond(status_code=500) + client = httpx.Client(base_url="https://mycantaloupe.com") + with pytest.raises(httpx.HTTPStatusError): + _get_anti_forgery_state(client) + + +# ───────────────────────────────────────────────── +# _do_login +# ───────────────────────────────────────────────── + + +@respx.mock +def test_do_login_success_with_result(): + respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {"UserId": 42}}) + client = httpx.Client(base_url="https://mycantaloupe.com") + result = _do_login(client, "u@test.com", "pw", "csrf-token") + assert result["SuccessfulResult"]["UserId"] == 42 + + +@respx.mock +def test_do_login_success_with_redirect(): + respx.post(SIGNIN_API).respond(json={"RedirectUrl": "/cs4/VueMachineList"}) + client = httpx.Client(base_url="https://mycantaloupe.com") + result = _do_login(client, "u@test.com", "pw", "csrf-token") + assert result["RedirectUrl"] == "/cs4/VueMachineList" + + +@respx.mock +def test_do_login_error_message(): + respx.post(SIGNIN_API).respond( + json={"ErrorMessage": "Incorrect username or password"} + ) + client = httpx.Client(base_url="https://mycantaloupe.com") + with pytest.raises(AuthenticationError, match="Login rejected: Incorrect"): + _do_login(client, "bad@test.com", "wrong", "csrf-token") + + +@respx.mock +def test_do_login_not_verified(): + respx.post(SIGNIN_API).respond( + json={"NotVerifiedUserEmail": "u@test.com"} + ) + client = httpx.Client(base_url="https://mycantaloupe.com") + with pytest.raises(AuthenticationError, match="Email not verified"): + _do_login(client, "u@test.com", "pw", "csrf-token") + + +@respx.mock +def test_do_login_non_200(): + respx.post(SIGNIN_API).respond(status_code=500, text="Internal Server Error") + client = httpx.Client(base_url="https://mycantaloupe.com") + with pytest.raises(AuthenticationError, match="failed with status 500"): + _do_login(client, "u@test.com", "pw", "csrf-token") + + +@respx.mock +def test_do_login_sends_correct_headers(): + route = respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}}) + client = httpx.Client(base_url="https://mycantaloupe.com") + _do_login(client, "u@test.com", "pw", "csrf-token-abc") + req = route.calls.last.request + assert req.headers["__RequestVerificationToken"] == "csrf-token-abc" + assert req.headers["Content-Type"] == "application/json" + body = req.content.decode() + assert "u@test.com" in body + assert '"pw"' in body + assert "FormsAuthReturnUrl" in body + + +@respx.mock +def test_do_login_unclear_response_still_returns(): + respx.post(SIGNIN_API).respond(json={"SomeField": "value"}) + client = httpx.Client(base_url="https://mycantaloupe.com") + result = _do_login(client, "u@test.com", "pw", "csrf-token") + assert result == {"SomeField": "value"} + + +# ───────────────────────────────────────────────── +# login() — full flow +# ───────────────────────────────────────────────── + + +@respx.mock +def test_login_full_flow(): + csrf_token = "CsrfToken123" + session_cookie = "abc123session" + auth_cookie = "authcookie456" + + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": csrf_token} + ) + login_route = respx.post(SIGNIN_API).respond( + json={"SuccessfulResult": {"UserId": 1}}, + headers=[ + ("Set-Cookie", f"ASP.NET_SessionId={session_cookie}; Path=/; HttpOnly; SameSite=Lax"), + ("Set-Cookie", f"ApplicationGatewayAffinity={auth_cookie}; Path=/"), + ], + ) + + with patch.dict( + os.environ, + {"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"}, + ): + auth = login() + + assert auth.email == "u@test.com" + assert auth.password == "pw" + assert "ASP.NET_SessionId" in auth.cookies + assert "ApplicationGatewayAffinity" in auth.cookies + assert auth.client.cookies["ApplicationGatewayAffinity"] == auth_cookie + + csrf_call = respx.get(ANTI_FORGERY_API).calls.last + assert csrf_call is not None + login_call = login_route.calls.last + assert login_call.request.headers["__RequestVerificationToken"] == csrf_token + + +@respx.mock +def test_login_with_explicit_credentials(): + csrf_token = "token" + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": csrf_token} + ) + respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}}) + + auth = login(email="explicit@test.com", password="explicitpw") + assert auth.email == "explicit@test.com" + assert auth.password == "explicitpw" + + +@respx.mock +def test_login_with_custom_client(): + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": "t"} + ) + respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}}) + + custom_client = httpx.Client(base_url="https://mycantaloupe.com", timeout=30.0) + auth = login(email="u@test.com", password="pw", client=custom_client) + assert auth.client is custom_client + + +def test_login_missing_credentials(): + with patch.dict(os.environ, {}, clear=True): + with pytest.raises(AuthenticationError, match="Credentials not found"): + login() + + +# ───────────────────────────────────────────────── +# get_session() — convenience factory +# ───────────────────────────────────────────────── + + +@respx.mock +def test_get_session_force_login(): + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": "t"} + ) + respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}}) + + with patch.dict( + os.environ, + {"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"}, + ): + client = get_session(force_login=True) + + assert isinstance(client, httpx.Client) + assert respx.get(ANTI_FORGERY_API).called + assert respx.post(SIGNIN_API).called + + +@respx.mock +def test_get_session_cached(): + respx.get("https://mycantaloupe.com/cs4/VueMachineList").respond( + status_code=200, + headers={ + "Set-Cookie": "ApplicationGatewayAffinity=cached-cookie; Path=/" + }, + ) + + with patch.dict( + os.environ, + {"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"}, + ): + client = get_session() + + assert isinstance(client, httpx.Client) + assert client.cookies.get("ApplicationGatewayAffinity") == "cached-cookie" + assert not respx.get(ANTI_FORGERY_API).called + assert not respx.post(SIGNIN_API).called + + +@respx.mock +def test_get_session_probe_fails_triggers_login(): + respx.get("https://mycantaloupe.com/cs4/VueMachineList").respond(status_code=200) + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": "t"} + ) + respx.post(SIGNIN_API).respond(json={"SuccessfulResult": {}}) + + with patch.dict( + os.environ, + {"CANTALOUPE_EMAIL": "u@test.com", "CANTALOUPE_PASSWORD": "pw"}, + ): + client = get_session() + + assert isinstance(client, httpx.Client) + assert respx.post(SIGNIN_API).called + + +# ───────────────────────────────────────────────── +# AuthSession.refresh() +# ───────────────────────────────────────────────── + + +@respx.mock +def test_auth_session_refresh_cookies_present(): + client = httpx.Client(base_url="https://mycantaloupe.com") + client.cookies.set("ApplicationGatewayAffinity", "still-here", domain="mycantaloupe.com") + session = AuthSession(client=client, email="u@test.com", password="pw") + assert session.refresh() is True + + +@respx.mock +def test_auth_session_refresh_relogin_success(): + client = httpx.Client(base_url="https://mycantaloupe.com") + session = AuthSession(client=client, email="u@test.com", password="pw") + + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": "new-token"} + ) + respx.post(SIGNIN_API).respond( + json={"SuccessfulResult": {}}, + headers={ + "Set-Cookie": "ApplicationGatewayAffinity=new-auth; Path=/" + }, + ) + + assert session.refresh() is True + assert session.client.cookies.get("ApplicationGatewayAffinity") == "new-auth" + + +@respx.mock +def test_auth_session_refresh_relogin_fails(): + client = httpx.Client(base_url="https://mycantaloupe.com") + session = AuthSession(client=client, email="u@test.com", password="bad") + + respx.get(ANTI_FORGERY_API).respond( + json={"HeaderName": "__RequestVerificationToken", "Token": "t"} + ) + respx.post(SIGNIN_API).respond( + json={"ErrorMessage": "Invalid credentials"} + ) + + assert session.refresh() is False + + +# ───────────────────────────────────────────────── +# Constants +# ───────────────────────────────────────────────── + + +def test_auth_cookie_names(): + assert "ApplicationGatewayAffinity" in AUTH_COOKIE_NAMES + assert "AeonPrincipalCacheVersion" in AUTH_COOKIE_NAMES + + +def test_url_constants(): + assert SIGNIN_API == "https://mycantaloupe.com/login/api/SignIn/SignIn/en-US" + assert ANTI_FORGERY_API == "https://mycantaloupe.com/login/api/DataContext/GenerateAntiForgeryState"