diff --git a/auth.py b/auth.py new file mode 100644 index 0000000..a669357 --- /dev/null +++ b/auth.py @@ -0,0 +1,116 @@ +# Native modules +import logging +import os +from typing import Dict, Any, Optional + +# Third-party modules +import requests +from urllib.parse import urlencode + +# Configuration constants (will be moved from crossfit_booker.py) +APPLICATION_ID = "81560887" +DEVICE_TYPE = "3" +APP_VERSION = "5.09.21" + +class AuthHandler: + """ + A class for handling authentication with the CrossFit booking system. + This class is responsible for performing login, retrieving auth tokens, + and providing authentication headers. + """ + + def __init__(self, username: str, password: str) -> None: + """ + Initialize the AuthHandler with credentials. + + Args: + username (str): The username for authentication. + password (str): The password for authentication. + """ + self.username = username + self.password = password + self.auth_token: Optional[str] = None + self.user_id: Optional[str] = None + self.session = requests.Session() + self.base_headers: Dict[str, str] = { + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0", + "Content-Type": "application/x-www-form-urlencoded", + "Nubapp-Origin": "user_apps", + } + self.session.headers.update(self.base_headers) + + def get_auth_headers(self) -> Dict[str, str]: + """ + Return headers with authorization if available. + + Returns: + Dict[str, str]: Headers dictionary with authorization if available. + """ + headers: Dict[str, str] = self.base_headers.copy() + if self.auth_token: + headers["Authorization"] = f"Bearer {self.auth_token}" + return headers + + def login(self) -> bool: + """ + Authenticate and get the bearer token. + + Returns: + bool: True if login is successful, False otherwise. + """ + try: + # First login endpoint + login_params: Dict[str, str] = { + "app_version": APP_VERSION, + "device_type": DEVICE_TYPE, + "username": self.username, + "password": self.password + } + + response: requests.Response = self.session.post( + "https://sport.nubapp.com/api/v4/users/checkUser.php", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + data=urlencode(login_params)) + + if not response.ok: + logging.error(f"First login step failed: {response.status_code} - {response.text[:100]}") + return False + + try: + login_data: Dict[str, Any] = response.json() + self.user_id = str(login_data["data"]["user"]["id_user"]) + except (KeyError, ValueError) as e: + logging.error(f"Error during login: {str(e)} - Response: {response.text}") + return False + + # Second login endpoint + response: requests.Response = self.session.post( + "https://sport.nubapp.com/api/v4/login", + headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"}, + data=urlencode({ + "device_type": DEVICE_TYPE, + "username": self.username, + "password": self.password + })) + + if response.ok: + try: + login_data: Dict[str, Any] = response.json() + self.auth_token = login_data.get("token") + except (KeyError, ValueError) as e: + logging.error(f"Error during login: {str(e)} - Response: {response.text}") + return False + + if self.auth_token and self.user_id: + logging.info("Successfully logged in") + return True + else: + logging.error(f"Login failed: {response.status_code} - {response.text[:100]}") + return False + + except requests.exceptions.RequestException as e: + logging.error(f"Request error during login: {str(e)}") + return False + except Exception as e: + logging.error(f"Unexpected error during login: {str(e)}") + return False \ No newline at end of file diff --git a/book_crossfit.py b/book_crossfit.py index 9a3c727..bd4b4d4 100755 --- a/book_crossfit.py +++ b/book_crossfit.py @@ -31,5 +31,5 @@ if __name__ == "__main__": exit(1) # Start the continuous booking loop - asyncio.run(booker.run()) + booker.run() logging.info("Script completed") diff --git a/booker.py b/booker.py new file mode 100644 index 0000000..63f4787 --- /dev/null +++ b/booker.py @@ -0,0 +1,310 @@ +# Native modules +import logging +import traceback +import os +import time +from datetime import datetime, timedelta, date + +# Third-party modules +import requests +from dateutil.parser import parse +import pytz +from dotenv import load_dotenv +from urllib.parse import urlencode +from typing import List, Dict, Optional, Any, Tuple + +# Import the SessionNotifier class +from session_notifier import SessionNotifier + +# Import the preferred sessions from the session_config module +from session_config import PREFERRED_SESSIONS + +# Import the AuthHandler class +from auth import AuthHandler + +load_dotenv() + +# Configuration +USERNAME = os.environ.get("CROSSFIT_USERNAME") +PASSWORD = os.environ.get("CROSSFIT_PASSWORD") + +if not all([USERNAME, PASSWORD]): + raise ValueError("Missing environment variables: CROSSFIT_USERNAME and/or CROSSFIT_PASSWORD") + +APPLICATION_ID = "81560887" +CATEGORY_ID = "677" # Activity category ID for CrossFit +TIMEZONE = "Europe/Paris" # Adjust to your timezone +# Booking window configuration (can be overridden by environment variables) +# TARGET_RESERVATION_TIME: string "HH:MM" local time when bookings open (default 20:01) +# BOOKING_WINDOW_END_DELTA_MINUTES: int minutes after target time to stop booking (default 10) +TARGET_RESERVATION_TIME = os.environ.get("TARGET_RESERVATION_TIME", "20:01") +BOOKING_WINDOW_END_DELTA_MINUTES = int(os.environ.get("BOOKING_WINDOW_END_DELTA_MINUTES", "10")) +DEVICE_TYPE = "3" + +# Retry configuration +RETRY_MAX = 3 +RETRY_BACKOFF = 1 +APP_VERSION = "5.09.21" + +class Booker: + """ + A class for handling the main booking logic. + This class is designed to be used as a standalone component + that can be initialized with authentication and session management + and used to perform the booking process. + """ + + def __init__(self, auth_handler: AuthHandler, notifier: SessionNotifier) -> None: + """ + Initialize the Booker with necessary attributes. + + Args: + auth_handler (AuthHandler): AuthHandler instance for authentication. + notifier (SessionNotifier): SessionNotifier instance for sending notifications. + """ + self.auth_handler = auth_handler + self.notifier = notifier + + # Initialize the session and headers + self.session: requests.Session = requests.Session() + self.base_headers: Dict[str, str] = { + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0", + "Content-Type": "application/x-www-form-urlencoded", + "Nubapp-Origin": "user_apps", + } + self.session.headers.update(self.base_headers) + + # Define mandatory parameters for API calls + self.mandatory_params: Dict[str, str] = { + "app_version": APP_VERSION, + "device_type": DEVICE_TYPE, + "id_application": APPLICATION_ID, + "id_category_activity": CATEGORY_ID + } + + def get_auth_headers(self) -> Dict[str, str]: + """ + Return headers with authorization from the AuthHandler. + + Returns: + Dict[str, str]: Headers dictionary with authorization if available. + """ + return self.auth_handler.get_auth_headers() + + async def booker(self, current_time: datetime) -> None: + """ + Run one cycle of checking and booking sessions. + + Args: + current_time (datetime): Current time for comparison. + """ + # Calculate date range to check (current day, day + 1, and day + 2) + start_date: date = current_time.date() + end_date: date = start_date + timedelta(days=2) # Only go up to day + 2 + + # Get available sessions + sessions_data: Optional[Dict[str, Any]] = self.get_available_sessions(start_date, end_date) + if not sessions_data or not sessions_data.get("success", False): + logging.error("No sessions available or error fetching sessions") + return + + activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", []) + + # Display all available sessions within the date range + self.display_upcoming_sessions(activities, current_time) + + # Find sessions to book (preferred only) within allowed date range + found_preferred_sessions: List[Dict[str, Any]] = [] + + for session in activities: + session_time: datetime = parse(session["start_timestamp"]) + if not session_time.tzinfo: + session_time = pytz.timezone(TIMEZONE).localize(session_time) + + # Check if session is within allowed date range (current day, day + 1, or day + 2) + days_diff = (session_time.date() - current_time.date()).days + if not (0 <= days_diff <= 2): + continue # Skip sessions outside the allowed date range + + # Check if session is preferred and bookable + if self.is_session_bookable(session, current_time): + if self.matches_preferred_session(session, current_time): + found_preferred_sessions.append(session) + + # Display preferred sessions found + if found_preferred_sessions: + logging.info("Preferred sessions found:") + for session in found_preferred_sessions: + session_time: datetime = parse(session["start_timestamp"]) + if not session_time.tzinfo: + session_time = pytz.timezone(TIMEZONE).localize(session_time) + logging.info(f"ID: {session['id_activity_calendar']}, Name: {session['name_activity']}, Time: {session_time.strftime('%Y-%m-%d %H:%M')}") + else: + logging.info("No matching preferred sessions found") + + # Book preferred sessions + if not found_preferred_sessions: + logging.info("No matching sessions found to book") + return + + # Book sessions (preferred first) + sessions_to_book = [("Preferred", session) for session in found_preferred_sessions] + sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1) + booked_sessions = [] + + for session_type, session in sessions_to_book: + session_time: datetime = parse(session["start_timestamp"]) + logging.info(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})") + if self.book_session(session["id_activity_calendar"]): + # Display booked session + booked_sessions.append(session) + logging.info(f"Successfully booked {session_type} session at {session_time}") + + # Notify about booked session + session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}" + await self.notifier.notify_session_booking(session_details) + else: + logging.error(f"Failed to book {session_type} session at {session_time}") + + # Notify about failed booking + session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}" + await self.notifier.notify_impossible_booking(session_details) + logging.info(f"Notified about impossible booking for {session_type} session at {session_time}") + + # Display all booked session(s) + if booked_sessions: + logging.info("Booked sessions:") + for session in booked_sessions: + session_time: datetime = parse(session["start_timestamp"]) + if not session_time.tzinfo: + session_time = pytz.timezone(TIMEZONE).localize(session_time) + logging.info(f"ID: {session['id_activity_calendar']}, Name: {session['name_activity']}, Time: {session_time.strftime('%Y-%m-%d %H:%M')}") + else: + logging.info("No sessions were booked") + + async def run(self) -> None: + """ + Main execution loop. + """ + # Set up timezone + tz: pytz.timezone = pytz.timezone(TIMEZONE) + + # Parse TARGET_RESERVATION_TIME to get the target hour and minute + target_hour, target_minute = map(int, TARGET_RESERVATION_TIME.split(":")) + target_time = datetime.now(tz).replace(hour=target_hour, minute=target_minute, second=0, microsecond=0) + booking_window_end = target_time + timedelta(minutes=BOOKING_WINDOW_END_DELTA_MINUTES) + + # Initial login + if not self.auth_handler.login(): + logging.error("Authentication failed - exiting program") + return + + try: + while True: + try: + current_time: datetime = datetime.now(tz) + logging.info(f"Current time: {current_time}") + + # Only book sessions if current time is within the booking window + if target_time <= current_time <= booking_window_end: + # Run booking cycle to check for preferred sessions and book + await self.booker(current_time) + # Wait for a short time before next check + time.sleep(60) + else: + # Check again in 5 minutes if outside booking window + time.sleep(300) + except Exception as e: + logging.error(f"Unexpected error in booking cycle: {str(e)} - Traceback: {traceback.format_exc()}") + time.sleep(60) # Wait before retrying after error + except KeyboardInterrupt: + self.quit() + + def quit(self) -> None: + """ + Clean up resources and exit the script. + """ + logging.info("Script interrupted by user. Quitting...") + exit(0) + + def get_available_sessions(self, start_date: date, end_date: date) -> Optional[Dict[str, Any]]: + """ + Fetch available sessions from the API. + + Args: + start_date (date): Start date for fetching sessions. + end_date (date): End date for fetching sessions. + + Returns: + Optional[Dict[str, Any]]: Dictionary containing available sessions if successful, None otherwise. + """ + from session_manager import SessionManager + session_manager = SessionManager(self.auth_handler) + return session_manager.get_available_sessions(start_date, end_date) + + def book_session(self, session_id: str) -> bool: + """ + Book a specific session. + + Args: + session_id (str): ID of the session to book. + + Returns: + bool: True if booking is successful, False otherwise. + """ + from session_manager import SessionManager + session_manager = SessionManager(self.auth_handler) + return session_manager.book_session(session_id) + + def get_booked_sessions(self) -> List[Dict[str, Any]]: + """ + Get a list of booked sessions. + + Returns: + A list of dictionaries containing information about the booked sessions. + """ + from session_manager import SessionManager + session_manager = SessionManager(self.auth_handler) + return session_manager.get_booked_sessions() + + def is_session_bookable(self, session: Dict[str, Any], current_time: datetime) -> bool: + """ + Check if a session is bookable based on user_info. + + Args: + session (Dict[str, Any]): Session data. + current_time (datetime): Current time for comparison. + + Returns: + bool: True if the session is bookable, False otherwise. + """ + from session_manager import SessionManager + session_manager = SessionManager(self.auth_handler) + return session_manager.is_session_bookable(session, current_time) + + def matches_preferred_session(self, session: Dict[str, Any], current_time: datetime) -> bool: + """ + Check if session matches one of your preferred sessions with exact matching. + + Args: + session (Dict[str, Any]): Session data. + current_time (datetime): Current time for comparison. + + Returns: + bool: True if the session matches a preferred session, False otherwise. + """ + from session_manager import SessionManager + session_manager = SessionManager(self.auth_handler) + return session_manager.matches_preferred_session(session, current_time) + + def display_upcoming_sessions(self, sessions: List[Dict[str, Any]], current_time: datetime) -> None: + """ + Display upcoming sessions with ID, name, date, and time. + + Args: + sessions (List[Dict[str, Any]]): List of session data. + current_time (datetime): Current time for comparison. + """ + from session_manager import SessionManager + session_manager = SessionManager(self.auth_handler) + session_manager.display_upcoming_sessions(sessions, current_time) \ No newline at end of file diff --git a/crossfit_booker.py b/crossfit_booker.py index d953fd0..ce6f7b6 100644 --- a/crossfit_booker.py +++ b/crossfit_booker.py @@ -1,25 +1,22 @@ # Native modules import logging -import traceback import os -import time -import difflib -from datetime import datetime, timedelta, date +from typing import Dict, Any, Optional -# Third-party modules -import requests -from dateutil.parser import parse -import pytz -from dotenv import load_dotenv -from urllib.parse import urlencode -from typing import List, Dict, Optional, Any, Tuple +# Import the AuthHandler class +from auth import AuthHandler + +# Import the SessionManager class +from session_manager import SessionManager + +# Import the Booker class +from booker import Booker # Import the SessionNotifier class from session_notifier import SessionNotifier -# Import the preferred sessions from the session_config module -from session_config import PREFERRED_SESSIONS - +# Load environment variables +from dotenv import load_dotenv load_dotenv() # Configuration @@ -29,52 +26,23 @@ PASSWORD = os.environ.get("CROSSFIT_PASSWORD") if not all([USERNAME, PASSWORD]): raise ValueError("Missing environment variables: CROSSFIT_USERNAME and/or CROSSFIT_PASSWORD") -APPLICATION_ID = "81560887" -CATEGORY_ID = "677" # Activity category ID for CrossFit -TIMEZONE = "Europe/Paris" # Adjust to your timezone -# Booking window configuration (can be overridden by environment variables) -# TARGET_RESERVATION_TIME: string "HH:MM" local time when bookings open (default 20:01) -# BOOKING_WINDOW_END_DELTA_MINUTES: int minutes after target time to stop booking (default 10) -TARGET_RESERVATION_TIME = os.environ.get("TARGET_RESERVATION_TIME", "20:01") -BOOKING_WINDOW_END_DELTA_MINUTES = int(os.environ.get("BOOKING_WINDOW_END_DELTA_MINUTES", "10")) -DEVICE_TYPE = "3" - -# Retry configuration -RETRY_MAX = 3 -RETRY_BACKOFF = 1 -APP_VERSION = "5.09.21" - - class CrossFitBooker: """ - A class for automating the booking of CrossFit sessions. - This class handles authentication, session availability checking, - booking, and notifications for CrossFit sessions. + A simple orchestrator class for the CrossFit booking system. + + This class is responsible for initializing and coordinating the other components + (AuthHandler, SessionManager, and Booker) but does not implement the actual functionality. """ def __init__(self) -> None: """ - Initialize the CrossFitBooker with necessary attributes. - Sets up authentication tokens, session headers, mandatory parameters, - and initializes the SessionNotifier for sending notifications. + Initialize the CrossFitBooker with necessary components. """ - self.auth_token: Optional[str] = None - self.user_id: Optional[str] = None - self.session: requests.Session = requests.Session() - self.base_headers: Dict[str, str] = { - "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0", - "Content-Type": "application/x-www-form-urlencoded", - "Nubapp-Origin": "user_apps", - } - self.session.headers.update(self.base_headers) + # Initialize the AuthHandler with credentials from environment variables + self.auth_handler = AuthHandler(USERNAME, PASSWORD) - # Define mandatory parameters for API calls - self.mandatory_params: Dict[str, str] = { - "app_version": APP_VERSION, - "device_type": DEVICE_TYPE, - "id_application": APPLICATION_ID, - "id_category_activity": CATEGORY_ID - } + # Initialize the SessionManager with the AuthHandler + self.session_manager = SessionManager(self.auth_handler) # Initialize the SessionNotifier with credentials from environment variables email_credentials = { @@ -99,441 +67,14 @@ class CrossFitBooker: enable_telegram=enable_telegram ) - def get_auth_headers(self) -> Dict[str, str]: + # Initialize the Booker with the AuthHandler and SessionNotifier + self.booker = Booker(self.auth_handler, self.notifier) + + def run(self) -> None: """ - Return headers with authorization if available. - Returns: - Dict[str, str]: Headers dictionary with authorization if available. + Start the booking process. + + This method initiates the booking process by running the Booker's main execution loop. """ - headers: Dict[str, str] = self.base_headers.copy() - if self.auth_token: - headers["Authorization"] = f"Bearer {self.auth_token}" - return headers - - def login(self) -> bool: - """ - Authenticate and get the bearer token. - Returns: - bool: True if login is successful, False otherwise. - """ - try: - # First login endpoint - login_params: Dict[str, str] = { - "app_version": APP_VERSION, - "device_type": DEVICE_TYPE, - "username": USERNAME, - "password": PASSWORD - } - - response: requests.Response = self.session.post( - "https://sport.nubapp.com/api/v4/users/checkUser.php", - headers={"Content-Type": "application/x-www-form-urlencoded"}, - data=urlencode(login_params)) - - if not response.ok: - logging.error(f"First login step failed: {response.status_code} - {response.text[:100]}") - return False - - try: - login_data: Dict[str, Any] = response.json() - self.user_id = str(login_data["data"]["user"]["id_user"]) - except (KeyError, ValueError) as e: - logging.error(f"Error during login: {str(e)} - Response: {response.text}") - return False - - # Second login endpoint - response: requests.Response = self.session.post( - "https://sport.nubapp.com/api/v4/login", - headers={"Content-Type": "application/x-www-form-urlencoded;charset=UTF-8"}, - data=urlencode({ - "device_type": DEVICE_TYPE, - "username": USERNAME, - "password": PASSWORD - })) - - if response.ok: - try: - login_data: Dict[str, Any] = response.json() - self.auth_token = login_data.get("token") - except (KeyError, ValueError) as e: - logging.error(f"Error during login: {str(e)} - Response: {response.text}") - return False - - if self.auth_token and self.user_id: - logging.info("Successfully logged in") - return True - else: - logging.error(f"Login failed: {response.status_code} - {response.text[:100]}") - return False - - except requests.exceptions.RequestException as e: - logging.error(f"Request error during login: {str(e)}") - return False - except Exception as e: - logging.error(f"Unexpected error during login: {str(e)}") - return False - - def get_available_sessions(self, start_date: date, end_date: date) -> Optional[Dict[str, Any]]: - """ - Fetch available sessions from the API. - Args: - start_date (date): Start date for fetching sessions. - end_date (date): End date for fetching sessions. - Returns: - Optional[Dict[str, Any]]: Dictionary containing available sessions if successful, None otherwise. - """ - if not self.auth_token or not self.user_id: - logging.error("Authentication required - missing token or user ID") - return None - - url: str = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php" - - # Prepare request with mandatory parameters - request_data: Dict[str, str] = self.mandatory_params.copy() - request_data.update({ - "id_user": self.user_id, - "start_timestamp": start_date.strftime("%d-%m-%Y"), - "end_timestamp": end_date.strftime("%d-%m-%Y") - }) - - # Add retry logic - for retry in range(RETRY_MAX): - try: - response: requests.Response = self.session.post( - url, - headers=self.get_auth_headers(), - data=urlencode(request_data), - timeout=10 - ) - - if response.status_code == 200: - return response.json() - elif response.status_code == 401: - logging.error("401 Unauthorized - token may be expired or invalid") - return None - elif 500 <= response.status_code < 600: - logging.error(f"Server error {response.status_code}") - raise requests.exceptions.ConnectionError(f"Server error {response.status_code}") - else: - logging.error(f"Unexpected status code: {response.status_code} - {response.text[:100]}") - return None - - except requests.exceptions.RequestException as e: - if retry == RETRY_MAX - 1: - logging.error(f"Final retry failed: {str(e)}") - raise - wait_time: int = RETRY_BACKOFF * (2 ** retry) - logging.warning(f"Request failed (attempt {retry+1}/{RETRY_MAX}): {str(e)}. Retrying in {wait_time}s...") - time.sleep(wait_time) - - return None - - def book_session(self, session_id: str) -> bool: - """ - Book a specific session. - Args: - session_id (str): ID of the session to book. - Returns: - bool: True if booking is successful, False otherwise. - """ - url = "https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php" - data = { - **self.mandatory_params, - "id_activity_calendar": session_id, - "id_user": self.user_id, - "action_by": self.user_id, - "n_guests": "0", - "booked_on": "1", - "device_type": self.mandatory_params["device_type"], - "token": self.auth_token - } - - for retry in range(RETRY_MAX): - try: - response: requests.Response = self.session.post( - url, - headers=self.get_auth_headers(), - data=urlencode(data), - timeout=10 - ) - - if response.status_code == 200: - json_response: Dict[str, Any] = response.json() - if json_response.get("success", False): - logging.info(f"Successfully booked session {session_id}") - return True - else: - logging.error(f"API returned success:false: {json_response} - Session ID: {session_id}") - return False - - logging.error(f"HTTP {response.status_code}: {response.text[:100]}") - return False - - except requests.exceptions.RequestException as e: - if retry == RETRY_MAX - 1: - logging.error(f"Final retry failed: {str(e)}") - raise - wait_time: int = RETRY_BACKOFF * (2 ** retry) - logging.warning(f"Request failed (attempt {retry+1}/{RETRY_MAX}): {str(e)}. Retrying in {wait_time}s...") - time.sleep(wait_time) - - logging.error(f"Failed to complete request after {RETRY_MAX} attempts") - return False - - def get_booked_sessions(self) -> List[Dict[str, Any]]: - """ - Get a list of booked sessions. - - Returns: - A list of dictionaries containing information about the booked sessions. - """ - url = "https://sport.nubapp.com/api/v4/activities/getBookedActivities.php" - data = { - **self.mandatory_params, - "id_user": self.user_id, - "action_by": self.user_id - } - - for retry in range(RETRY_MAX): - try: - response: requests.Response = self.session.post( - url, - headers=self.get_auth_headers(), - data=urlencode(data), - timeout=10 - ) - - if response.status_code == 200: - json_response: Dict[str, Any] = response.json() - if json_response.get("success", False): - return json_response.get("data", []) - logging.error(f"API returned success:false: {json_response}") - return [] - - logging.error(f"HTTP {response.status_code}: {response.text[:100]}") - return [] - - except requests.exceptions.RequestException as e: - if retry == RETRY_MAX - 1: - logging.error(f"Final retry failed: {str(e)}") - raise - wait_time: int = RETRY_BACKOFF * (2 ** retry) - logging.warning(f"Request failed (attempt {retry+1}/{RETRY_MAX}): {str(e)}. Retrying in {wait_time}s...") - time.sleep(wait_time) - - logging.error(f"Failed to complete request after {RETRY_MAX} attempts") - return [] - - def is_session_bookable(self, session: Dict[str, Any], current_time: datetime) -> bool: - """ - Check if a session is bookable based on user_info. - Args: - session (Dict[str, Any]): Session data. - current_time (datetime): Current time for comparison. - Returns: - bool: True if the session is bookable, False otherwise. - """ - user_info: Dict[str, Any] = session.get("user_info", {}) - - # First check if can_join is true (primary condition) - if user_info.get("can_join", False): - return True - - # Default case: not bookable - return False - - def matches_preferred_session(self, session: Dict[str, Any], current_time: datetime) -> bool: - """ - Check if session matches one of your preferred sessions with exact matching. - Args: - session (Dict[str, Any]): Session data. - current_time (datetime): Current time for comparison. - Returns: - bool: True if the session matches a preferred session, False otherwise. - """ - try: - session_time: datetime = parse(session["start_timestamp"]) - if not session_time.tzinfo: - session_time = pytz.timezone(TIMEZONE).localize(session_time) - - day_of_week: int = session_time.weekday() - session_time_str: str = session_time.strftime("%H:%M") - session_name: str = session.get("name_activity", "").upper() - - for preferred_day, preferred_time, preferred_name in PREFERRED_SESSIONS: - # Exact match - if (day_of_week == preferred_day and - session_time_str == preferred_time and - preferred_name in session_name): - return True - - return False - - except Exception as e: - logging.error(f"Failed to check session: {str(e)} - Session: {session}") - return False - - def display_upcoming_sessions(self, sessions: List[Dict[str, Any]], current_time: datetime) -> None: - """ - Display upcoming sessions with ID, name, date, and time. - - Args: - sessions (List[Dict[str, Any]]): List of session data. - current_time (datetime): Current time for comparison. - """ - if not sessions: - logging.info("No sessions to display") - return - - logging.info("Upcoming sessions:") - logging.info("ID\t\tName\t\tDate\t\tTime") - logging.info("="*50) - - for session in sessions: - session_time: datetime = parse(session["start_timestamp"]) - if not session_time.tzinfo: - session_time = pytz.timezone(TIMEZONE).localize(session_time) - - # Format session details - session_id = session.get("id_activity_calendar", "N/A") - session_name = session.get("name_activity", "N/A") - session_date = session_time.strftime("%Y-%m-%d") - session_time_str = session_time.strftime("%H:%M") - - # Display session details - logging.info(f"{session_id}\t{session_name}\t{session_date}\t{session_time_str}") - - async def booker(self, current_time: datetime) -> None: - """ - Run one cycle of checking and booking sessions. - Args: - current_time (datetime): Current time for comparison. - """ - # Calculate date range to check (current day, day + 1, and day + 2) - start_date: date = current_time.date() - end_date: date = start_date + timedelta(days=2) # Only go up to day + 2 - - # Get available sessions - sessions_data: Optional[Dict[str, Any]] = self.get_available_sessions(start_date, end_date) - if not sessions_data or not sessions_data.get("success", False): - logging.error("No sessions available or error fetching sessions") - return - - activities: List[Dict[str, Any]] = sessions_data.get("data", {}).get("activities_calendar", []) - - # Display all available sessions within the date range - self.display_upcoming_sessions(activities, current_time) - - # Find sessions to book (preferred only) within allowed date range - found_preferred_sessions: List[Dict[str, Any]] = [] - - for session in activities: - session_time: datetime = parse(session["start_timestamp"]) - if not session_time.tzinfo: - session_time = pytz.timezone(TIMEZONE).localize(session_time) - - # Check if session is within allowed date range (current day, day + 1, or day + 2) - days_diff = (session_time.date() - current_time.date()).days - if not (0 <= days_diff <= 2): - continue # Skip sessions outside the allowed date range - - # Check if session is preferred and bookable - if self.is_session_bookable(session, current_time): - if self.matches_preferred_session(session, current_time): - found_preferred_sessions.append(session) - - # Display preferred sessions found - if found_preferred_sessions: - logging.info("Preferred sessions found:") - for session in found_preferred_sessions: - session_time: datetime = parse(session["start_timestamp"]) - if not session_time.tzinfo: - session_time = pytz.timezone(TIMEZONE).localize(session_time) - logging.info(f"ID: {session['id_activity_calendar']}, Name: {session['name_activity']}, Time: {session_time.strftime('%Y-%m-%d %H:%M')}") - else: - logging.info("No matching preferred sessions found") - - # Book preferred sessions - if not found_preferred_sessions: - logging.info("No matching sessions found to book") - return - - # Book sessions (preferred first) - sessions_to_book = [("Preferred", session) for session in found_preferred_sessions] - sessions_to_book.sort(key=lambda x: 0 if x[0] == "Preferred" else 1) - booked_sessions = [] - - for session_type, session in sessions_to_book: - session_time: datetime = parse(session["start_timestamp"]) - logging.info(f"Attempting to book {session_type} session at {session_time} ({session['name_activity']})") - if self.book_session(session["id_activity_calendar"]): - # Display booked session - booked_sessions.append(session) - logging.info(f"Successfully booked {session_type} session at {session_time}") - - # Notify about booked session - session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}" - await self.notifier.notify_session_booking(session_details) - else: - logging.error(f"Failed to book {session_type} session at {session_time}") - - # Notify about failed booking - session_details = f"{session['name_activity']} at {session_time.strftime('%Y-%m-%d %H:%M')}" - await self.notifier.notify_impossible_booking(session_details) - logging.info(f"Notified about impossible booking for {session_type} session at {session_time}") - - # Display all booked session(s) - if booked_sessions: - logging.info("Booked sessions:") - for session in booked_sessions: - session_time: datetime = parse(session["start_timestamp"]) - if not session_time.tzinfo: - session_time = pytz.timezone(TIMEZONE).localize(session_time) - logging.info(f"ID: {session['id_activity_calendar']}, Name: {session['name_activity']}, Time: {session_time.strftime('%Y-%m-%d %H:%M')}") - else: - logging.info("No sessions were booked") - - async def run(self) -> None: - """ - Main execution loop. - """ - # Set up timezone - tz: pytz.timezone = pytz.timezone(TIMEZONE) - - # Parse TARGET_RESERVATION_TIME to get the target hour and minute - target_hour, target_minute = map(int, TARGET_RESERVATION_TIME.split(":")) - target_time = datetime.now(tz).replace(hour=target_hour, minute=target_minute, second=0, microsecond=0) - booking_window_end = target_time + timedelta(minutes=BOOKING_WINDOW_END_DELTA_MINUTES) - - # Initial login - if not self.login(): - logging.error("Authentication failed - exiting program") - return - - try: - while True: - try: - current_time: datetime = datetime.now(tz) - logging.info(f"Current time: {current_time}") - - # Only book sessions if current time is within the booking window - if target_time <= current_time <= booking_window_end: - # Run booking cycle to check for preferred sessions and book - await self.booker(current_time) - # Wait for a short time before next check - time.sleep(60) - else: - # Check again in 5 minutes if outside booking window - time.sleep(300) - except Exception as e: - logging.error(f"Unexpected error in booking cycle: {str(e)} - Traceback: {traceback.format_exc()}") - time.sleep(60) # Wait before retrying after error - except KeyboardInterrupt: - self.quit() - - def quit(self) -> None: - """ - Clean up resources and exit the script. - """ - logging.info("Script interrupted by user. Quitting...") - exit(0) + import asyncio + asyncio.run(self.booker.run()) diff --git a/main.py b/main.py new file mode 100755 index 0000000..c1e67a1 --- /dev/null +++ b/main.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +""" +Main entry point for the CrossFit Booker application. +This script initializes the CrossFitBooker and starts the booking process. +""" + +import asyncio +import logging +from crossfit_booker import CrossFitBooker + +def main(): + """ + Main function to initialize the CrossFitBooker and start the booking process. + """ + # Set up logging + logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(levelname)s - %(message)s', + handlers=[ + logging.StreamHandler() + ] + ) + + # Initialize the CrossFitBooker + booker = CrossFitBooker() + + # Run the booking process + booker.run() + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/session_manager.py b/session_manager.py new file mode 100644 index 0000000..18d800a --- /dev/null +++ b/session_manager.py @@ -0,0 +1,295 @@ +# Native modules +import logging +import pytz +import time +from datetime import date +from typing import List, Dict, Optional, Any +from datetime import datetime, timedelta, date + +# Third-party modules +import requests +from dateutil.parser import parse + +# Import the preferred sessions from the session_config module +from session_config import PREFERRED_SESSIONS + +# Import the AuthHandler class +from auth import AuthHandler + +class SessionManager: + """ + A class for managing CrossFit sessions. + This class handles session availability checking, booking, + and session-related operations. + """ + + def __init__(self, auth_handler: AuthHandler) -> None: + """ + Initialize the SessionManager with necessary attributes. + + Args: + auth_handler (AuthHandler): AuthHandler instance for authentication. + """ + self.auth_handler = auth_handler + self.session = requests.Session() + self.base_headers = { + "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0", + "Content-Type": "application/x-www-form-urlencoded", + "Nubapp-Origin": "user_apps", + } + self.session.headers.update(self.base_headers) + + # Define mandatory parameters for API calls + self.mandatory_params = { + "app_version": "5.09.21", + "device_type": "3", + "id_application": "81560887", + "id_category_activity": "677" + } + + def get_auth_headers(self) -> Dict[str, str]: + """ + Return headers with authorization from the AuthHandler. + + Returns: + Dict[str, str]: Headers dictionary with authorization if available. + """ + return self.auth_handler.get_auth_headers() + + def get_available_sessions(self, start_date: date, end_date: date) -> Optional[Dict[str, Any]]: + """ + Fetch available sessions from the API. + + Args: + start_date (date): Start date for fetching sessions. + end_date (date): End date for fetching sessions. + + Returns: + Optional[Dict[str, Any]]: Dictionary containing available sessions if successful, None otherwise. + """ + if not self.auth_handler.auth_token or not self.auth_handler.user_id: + logging.error("Authentication required - missing token or user ID") + return None + + url = "https://sport.nubapp.com/api/v4/activities/getActivitiesCalendar.php" + + # Prepare request with mandatory parameters + request_data = self.mandatory_params.copy() + request_data.update({ + "id_user": self.auth_handler.user_id, + "start_timestamp": start_date.strftime("%d-%m-%Y"), + "end_timestamp": end_date.strftime("%d-%m-%Y") + }) + + # Add retry logic + for retry in range(3): + try: + response = self.session.post( + url, + headers=self.get_auth_headers(), + data=request_data, + timeout=10 + ) + + if response.status_code == 200: + return response.json() + elif response.status_code == 401: + logging.error("401 Unauthorized - token may be expired or invalid") + return None + elif 500 <= response.status_code < 600: + logging.error(f"Server error {response.status_code}") + raise requests.exceptions.ConnectionError(f"Server error {response.status_code}") + else: + logging.error(f"Unexpected status code: {response.status_code} - {response.text[:100]}") + return None + + except requests.exceptions.RequestException as e: + if retry == 2: + logging.error(f"Final retry failed: {str(e)}") + raise + wait_time = 1 * (2 ** retry) + logging.warning(f"Request failed (attempt {retry+1}/3): {str(e)}. Retrying in {wait_time}s...") + time.sleep(wait_time) + + return None + + def book_session(self, session_id: str) -> bool: + """ + Book a specific session. + + Args: + session_id (str): ID of the session to book. + + Returns: + bool: True if booking is successful, False otherwise. + """ + url = "https://sport.nubapp.com/api/v4/activities/bookActivityCalendar.php" + data = { + **self.mandatory_params, + "id_activity_calendar": session_id, + "id_user": self.auth_handler.user_id, + "action_by": self.auth_handler.user_id, + "n_guests": "0", + "booked_on": "1", + "device_type": self.mandatory_params["device_type"], + "token": self.auth_handler.auth_token + } + + for retry in range(3): + try: + response = self.session.post( + url, + headers=self.get_auth_headers(), + data=data, + timeout=10 + ) + + if response.status_code == 200: + json_response = response.json() + if json_response.get("success", False): + logging.info(f"Successfully booked session {session_id}") + return True + else: + logging.error(f"API returned success:false: {json_response} - Session ID: {session_id}") + return False + + logging.error(f"HTTP {response.status_code}: {response.text[:100]}") + return False + + except requests.exceptions.RequestException as e: + if retry == 2: + logging.error(f"Final retry failed: {str(e)}") + raise + wait_time = 1 * (2 ** retry) + logging.warning(f"Request failed (attempt {retry+1}/3): {str(e)}. Retrying in {wait_time}s...") + time.sleep(wait_time) + + logging.error(f"Failed to complete request after 3 attempts") + return False + + def get_booked_sessions(self) -> List[Dict[str, Any]]: + """ + Get a list of booked sessions. + + Returns: + A list of dictionaries containing information about the booked sessions. + """ + url = "https://sport.nubapp.com/api/v4/activities/getBookedActivities.php" + data = { + **self.mandatory_params, + "id_user": self.auth_handler.user_id, + "action_by": self.auth_handler.user_id + } + + for retry in range(3): + try: + response = self.session.post( + url, + headers=self.get_auth_headers(), + data=data, + timeout=10 + ) + + if response.status_code == 200: + json_response = response.json() + if json_response.get("success", False): + return json_response.get("data", []) + logging.error(f"API returned success:false: {json_response}") + return [] + + logging.error(f"HTTP {response.status_code}: {response.text[:100]}") + return [] + + except requests.exceptions.RequestException as e: + if retry == 2: + logging.error(f"Final retry failed: {str(e)}") + raise + wait_time = 1 * (2 ** retry) + logging.warning(f"Request failed (attempt {retry+1}/3): {str(e)}. Retrying in {wait_time}s...") + time.sleep(wait_time) + + logging.error(f"Failed to complete request after 3 attempts") + return [] + + def is_session_bookable(self, session: Dict[str, Any], current_time: datetime) -> bool: + """ + Check if a session is bookable based on user_info. + + Args: + session (Dict[str, Any]): Session data. + current_time (datetime): Current time for comparison. + + Returns: + bool: True if the session is bookable, False otherwise. + """ + user_info = session.get("user_info", {}) + + # First check if can_join is true (primary condition) + if user_info.get("can_join", False): + return True + + # Default case: not bookable + return False + + def matches_preferred_session(self, session: Dict[str, Any], current_time: datetime) -> bool: + """ + Check if session matches one of your preferred sessions with exact matching. + + Args: + session (Dict[str, Any]): Session data. + current_time (datetime): Current time for comparison. + + Returns: + bool: True if the session matches a preferred session, False otherwise. + """ + try: + session_time = parse(session["start_timestamp"]) + if not session_time.tzinfo: + session_time = pytz.timezone("Europe/Paris").localize(session_time) + + day_of_week = session_time.weekday() + session_time_str = session_time.strftime("%H:%M") + session_name = session.get("name_activity", "").upper() + + for preferred_day, preferred_time, preferred_name in PREFERRED_SESSIONS: + # Exact match + if (day_of_week == preferred_day and + session_time_str == preferred_time and + preferred_name in session_name): + return True + + return False + + except Exception as e: + logging.error(f"Failed to check session: {str(e)} - Session: {session}") + return False + + def display_upcoming_sessions(self, sessions: List[Dict[str, Any]], current_time: datetime) -> None: + """ + Display upcoming sessions with ID, name, date, and time. + + Args: + sessions (List[Dict[str, Any]]): List of session data. + current_time (datetime): Current time for comparison. + """ + if not sessions: + logging.info("No sessions to display") + return + + logging.info("Upcoming sessions:") + logging.info("ID\t\tName\t\tDate\t\tTime") + logging.info("="*50) + + for session in sessions: + session_time = parse(session["start_timestamp"]) + if not session_time.tzinfo: + session_time = pytz.timezone("Europe/Paris").localize(session_time) + + # Format session details + session_id = session.get("id_activity_calendar", "N/A") + session_name = session.get("name_activity", "N/A") + session_date = session_time.strftime("%Y-%m-%d") + session_time_str = session_time.strftime("%H:%M") + + # Display session details + logging.info(f"{session_id}\t{session_name}\t{session_date}\t{session_time_str}") \ No newline at end of file diff --git a/test/test_auth.py b/test/test_auth.py new file mode 100644 index 0000000..ce909e6 --- /dev/null +++ b/test/test_auth.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +""" +Unit tests for AuthHandler class +""" + +import pytest +import os +import sys +import requests +from unittest.mock import patch, Mock + +# Add the parent directory to the path +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from auth import AuthHandler + +class TestAuthHandlerAuthHeaders: + """Test cases for get_auth_headers method""" + + def test_get_auth_headers_without_token(self): + """Test headers without auth token""" + auth_handler = AuthHandler('test_user', 'test_pass') + headers = auth_handler.get_auth_headers() + assert "Authorization" not in headers + assert headers["User-Agent"] == "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0" + + def test_get_auth_headers_with_token(self): + """Test headers with auth token""" + auth_handler = AuthHandler('test_user', 'test_pass') + auth_handler.auth_token = "test_token_123" + headers = auth_handler.get_auth_headers() + assert headers["Authorization"] == "Bearer test_token_123" + +class TestAuthHandlerLogin: + """Test cases for login method""" + + @patch('requests.Session.post') + def test_login_success(self, mock_post): + """Test successful login flow""" + # Mock first login response + mock_response1 = Mock() + mock_response1.ok = True + mock_response1.json.return_value = { + "data": { + "user": { + "id_user": "12345" + } + } + } + + # Mock second login response + mock_response2 = Mock() + mock_response2.ok = True + mock_response2.json.return_value = { + "token": "test_bearer_token" + } + + mock_post.side_effect = [mock_response1, mock_response2] + + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() + + assert result is True + assert auth_handler.user_id == "12345" + assert auth_handler.auth_token == "test_bearer_token" + + @patch('requests.Session.post') + def test_login_first_step_failure(self, mock_post): + """Test login failure on first step""" + mock_response = Mock() + mock_response.ok = False + mock_response.status_code = 400 + mock_response.text = "Bad Request" + + mock_post.return_value = mock_response + + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() + + assert result is False + assert auth_handler.user_id is None + assert auth_handler.auth_token is None + + @patch('requests.Session.post') + def test_login_second_step_failure(self, mock_post): + """Test login failure on second step""" + # First response succeeds + mock_response1 = Mock() + mock_response1.ok = True + mock_response1.json.return_value = { + "data": { + "user": { + "id_user": "12345" + } + } + } + + # Second response fails + mock_response2 = Mock() + mock_response2.ok = False + mock_response2.status_code = 401 + + mock_post.side_effect = [mock_response1, mock_response2] + + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() + + assert result is False + + @patch('requests.Session.post') + def test_login_json_parsing_error(self, mock_post): + """Test login with JSON parsing error""" + mock_response = Mock() + mock_response.ok = True + mock_response.json.side_effect = ValueError("Invalid JSON") + + mock_post.return_value = mock_response + + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() + + assert result is False + + @patch('requests.Session.post') + def test_login_request_exception(self, mock_post): + """Test login with request exception""" + mock_post.side_effect = requests.exceptions.ConnectionError("Network error") + + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() + + assert result is False + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/test/test_crossfit_booker.py b/test/test_crossfit_booker.py index d735049..6f1bb22 100755 --- a/test/test_crossfit_booker.py +++ b/test/test_crossfit_booker.py @@ -189,8 +189,9 @@ def test_book_session(): # Create a CrossFitBooker instance booker = CrossFitBooker() - + # Login to get the authentication token + # The login method now uses the AuthHandler internally booker.login() # Get available sessions diff --git a/test/test_crossfit_booker_auth.py b/test/test_crossfit_booker_auth.py index 6978eda..0496c8f 100644 --- a/test/test_crossfit_booker_auth.py +++ b/test/test_crossfit_booker_auth.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 """ -Unit tests for CrossFitBooker authentication methods +Unit tests for CrossFitBooker authentication methods using AuthHandler """ import pytest @@ -13,11 +13,11 @@ from unittest.mock import patch, Mock sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from crossfit_booker import CrossFitBooker - +from auth import AuthHandler class TestCrossFitBookerAuthHeaders: """Test cases for get_auth_headers method""" - + def test_get_auth_headers_without_token(self): """Test headers without auth token""" with patch.dict(os.environ, { @@ -28,7 +28,7 @@ class TestCrossFitBookerAuthHeaders: headers = booker.get_auth_headers() assert "Authorization" not in headers assert headers["User-Agent"] == "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0" - + def test_get_auth_headers_with_token(self): """Test headers with auth token""" with patch.dict(os.environ, { @@ -36,14 +36,13 @@ class TestCrossFitBookerAuthHeaders: 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() - booker.auth_token = "test_token_123" + booker.auth_handler.auth_token = "test_token_123" headers = booker.get_auth_headers() assert headers["Authorization"] == "Bearer test_token_123" - class TestCrossFitBookerLogin: """Test cases for login method""" - + @patch('requests.Session.post') def test_login_success(self, mock_post): """Test successful login flow""" @@ -57,27 +56,27 @@ class TestCrossFitBookerLogin: } } } - + # Mock second login response mock_response2 = Mock() mock_response2.ok = True mock_response2.json.return_value = { "token": "test_bearer_token" } - + mock_post.side_effect = [mock_response1, mock_response2] - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - + assert result is True - assert booker.user_id == "12345" - assert booker.auth_token == "test_bearer_token" - + assert booker.auth_handler.user_id == "12345" + assert booker.auth_handler.auth_token == "test_bearer_token" + @patch('requests.Session.post') def test_login_first_step_failure(self, mock_post): """Test login failure on first step""" @@ -85,20 +84,20 @@ class TestCrossFitBookerLogin: mock_response.ok = False mock_response.status_code = 400 mock_response.text = "Bad Request" - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - + assert result is False - assert booker.user_id is None - assert booker.auth_token is None - + assert booker.auth_handler.user_id is None + assert booker.auth_handler.auth_token is None + @patch('requests.Session.post') def test_login_second_step_failure(self, mock_post): """Test login failure on second step""" @@ -112,55 +111,54 @@ class TestCrossFitBookerLogin: } } } - + # Second response fails mock_response2 = Mock() mock_response2.ok = False mock_response2.status_code = 401 - + mock_post.side_effect = [mock_response1, mock_response2] - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - + assert result is False - + @patch('requests.Session.post') def test_login_json_parsing_error(self, mock_post): """Test login with JSON parsing error""" mock_response = Mock() mock_response.ok = True mock_response.json.side_effect = ValueError("Invalid JSON") - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - + assert result is False - + @patch('requests.Session.post') def test_login_request_exception(self, mock_post): """Test login with request exception""" mock_post.side_effect = requests.exceptions.ConnectionError("Network error") - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - - assert result is False + assert result is False if __name__ == "__main__": pytest.main([__file__, "-v"]) \ No newline at end of file diff --git a/test/test_crossfit_booker_final.py b/test/test_crossfit_booker_final.py index 0dc29dd..c43c396 100644 --- a/test/test_crossfit_booker_final.py +++ b/test/test_crossfit_booker_final.py @@ -14,10 +14,9 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from crossfit_booker import CrossFitBooker - class TestCrossFitBookerInit: """Test cases for CrossFitBooker initialization""" - + def test_init_success(self): """Test successful initialization with all required env vars""" with patch.dict(os.environ, { @@ -30,11 +29,11 @@ class TestCrossFitBookerInit: 'TELEGRAM_CHAT_ID': '12345' }): booker = CrossFitBooker() - assert booker.auth_token is None - assert booker.user_id is None + assert booker.auth_handler.auth_token is None + assert booker.auth_handler.user_id is None assert booker.session is not None assert booker.notifier is not None - + def test_init_missing_credentials(self): """Test initialization fails with missing credentials""" with patch.dict(os.environ, {}, clear=True): @@ -42,7 +41,7 @@ class TestCrossFitBookerInit: CrossFitBooker() except ValueError as e: assert str(e) == "Missing environment variables" - + def test_init_partial_credentials(self): """Test initialization fails with partial credentials""" with patch.dict(os.environ, { @@ -54,10 +53,9 @@ class TestCrossFitBookerInit: except ValueError as e: assert str(e) == "Missing environment variables" - class TestCrossFitBookerAuthHeaders: """Test cases for get_auth_headers method""" - + def test_get_auth_headers_without_token(self): """Test headers without auth token""" with patch.dict(os.environ, { @@ -68,7 +66,7 @@ class TestCrossFitBookerAuthHeaders: headers = booker.get_auth_headers() assert "Authorization" not in headers assert headers["User-Agent"] == "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:140.0) Gecko/20100101 Firefox/140.0" - + def test_get_auth_headers_with_token(self): """Test headers with auth token""" with patch.dict(os.environ, { @@ -76,14 +74,13 @@ class TestCrossFitBookerAuthHeaders: 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() - booker.auth_token = "test_token_123" + booker.auth_handler.auth_token = "test_token_123" headers = booker.get_auth_headers() assert headers["Authorization"] == "Bearer test_token_123" - class TestCrossFitBookerLogin: """Test cases for login method""" - + @patch('requests.Session.post') def test_login_success(self, mock_post): """Test successful login flow""" @@ -97,27 +94,27 @@ class TestCrossFitBookerLogin: } } } - + # Mock second login response mock_response2 = Mock() mock_response2.ok = True mock_response2.json.return_value = { "token": "test_bearer_token" } - + mock_post.side_effect = [mock_response1, mock_response2] - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - + assert result is True - assert booker.user_id == "12345" - assert booker.auth_token == "test_bearer_token" - + assert booker.auth_handler.user_id == "12345" + assert booker.auth_handler.auth_token == "test_bearer_token" + @patch('requests.Session.post') def test_login_first_step_failure(self, mock_post): """Test login failure on first step""" @@ -125,56 +122,55 @@ class TestCrossFitBookerLogin: mock_response.ok = False mock_response.status_code = 400 mock_response.text = "Bad Request" - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - + assert result is False - assert booker.user_id is None - assert booker.auth_token is None - + assert booker.auth_handler.user_id is None + assert booker.auth_handler.auth_token is None + @patch('requests.Session.post') def test_login_json_parsing_error(self, mock_post): """Test login with JSON parsing error""" mock_response = Mock() mock_response.ok = True mock_response.json.side_effect = ValueError("Invalid JSON") - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - + assert result is False - + @patch('requests.Session.post') def test_login_request_exception(self, mock_post): """Test login with request exception""" mock_post.side_effect = requests.exceptions.ConnectionError("Network error") - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() result = booker.login() - - assert result is False + assert result is False class TestCrossFitBookerGetAvailableSessions: """Test cases for get_available_sessions method""" - + def test_get_available_sessions_no_auth(self): """Test get_available_sessions without authentication""" with patch.dict(os.environ, { @@ -184,7 +180,7 @@ class TestCrossFitBookerGetAvailableSessions: booker = CrossFitBooker() result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) assert result is None - + @patch('requests.Session.post') def test_get_available_sessions_success(self, mock_post): """Test successful get_available_sessions""" @@ -198,39 +194,39 @@ class TestCrossFitBookerGetAvailableSessions: ] } } - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() - booker.auth_token = "test_token" - booker.user_id = "12345" - + booker.auth_handler.auth_token = "test_token" + booker.auth_handler.user_id = "12345" + result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) - + assert result is not None assert result["success"] is True - + @patch('requests.Session.post') def test_get_available_sessions_failure(self, mock_post): """Test get_available_sessions with API failure""" mock_response = Mock() mock_response.status_code = 400 mock_response.text = "Bad Request" - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() - booker.auth_token = "test_token" - booker.user_id = "12345" - + booker.auth_handler.auth_token = "test_token" + booker.auth_handler.user_id = "12345" + result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) - + assert result is None \ No newline at end of file diff --git a/test/test_crossfit_booker_sessions.py b/test/test_crossfit_booker_sessions.py index 7fad7f7..807aedf 100644 --- a/test/test_crossfit_booker_sessions.py +++ b/test/test_crossfit_booker_sessions.py @@ -14,21 +14,23 @@ import pytz sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from crossfit_booker import CrossFitBooker - +from session_manager import SessionManager +from auth import AuthHandler class TestCrossFitBookerGetAvailableSessions: """Test cases for get_available_sessions method""" - + def test_get_available_sessions_no_auth(self): """Test get_available_sessions without authentication""" with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) + result = session_manager.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) assert result is None - + @patch('requests.Session.post') def test_get_available_sessions_success(self, mock_post): """Test successful get_available_sessions""" @@ -42,122 +44,126 @@ class TestCrossFitBookerGetAvailableSessions: ] } } - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - booker.auth_token = "test_token" - booker.user_id = "12345" - - result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) - + auth_handler = AuthHandler('test_user', 'test_pass') + auth_handler.auth_token = "test_token" + auth_handler.user_id = "12345" + session_manager = SessionManager(auth_handler) + + result = session_manager.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) + assert result is not None assert result["success"] is True - + @patch('requests.Session.post') def test_get_available_sessions_401_error(self, mock_post): """Test get_available_sessions with 401 error""" mock_response = Mock() mock_response.status_code = 401 - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): booker = CrossFitBooker() - booker.auth_token = "test_token" - booker.user_id = "12345" - - result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) - - assert result is None + booker.auth_handler.auth_token = "test_token" + booker.auth_handler.user_id = "12345" + result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) + + assert result is None class TestCrossFitBookerBookSession: """Test cases for book_session method""" - + def test_book_session_no_auth(self): """Test book_session without authentication""" with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - result = booker.book_session("session_123") + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) + result = session_manager.book_session("session_123") assert result is False - + @patch('requests.Session.post') def test_book_session_success(self, mock_post): """Test successful book_session""" mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = {"success": True} - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - booker.auth_token = "test_token" - booker.user_id = "12345" - - result = booker.book_session("session_123") - + auth_handler = AuthHandler('test_user', 'test_pass') + auth_handler.auth_token = "test_token" + auth_handler.user_id = "12345" + session_manager = SessionManager(auth_handler) + + result = session_manager.book_session("session_123") + assert result is True - + @patch('requests.Session.post') def test_book_session_api_failure(self, mock_post): """Test book_session with API failure""" mock_response = Mock() mock_response.status_code = 200 mock_response.json.return_value = {"success": False, "error": "Session full"} - + mock_post.return_value = mock_response - + with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - booker.auth_token = "test_token" - booker.user_id = "12345" - - result = booker.book_session("session_123") - - assert result is False + auth_handler = AuthHandler('test_user', 'test_pass') + auth_handler.auth_token = "test_token" + auth_handler.user_id = "12345" + session_manager = SessionManager(auth_handler) + result = session_manager.book_session("session_123") + + assert result is False class TestCrossFitBookerIsSessionBookable: """Test cases for is_session_bookable method""" - + def test_is_session_bookable_can_join_true(self): """Test session bookable with can_join=True""" with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) session = {"user_info": {"can_join": True}} current_time = datetime.now(pytz.timezone("Europe/Paris")) - - result = booker.is_session_bookable(session, current_time) + + result = session_manager.is_session_bookable(session, current_time) assert result is True - + def test_is_session_bookable_booking_window_past(self): """Test session bookable with booking window in past""" with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) session = { "user_info": { "can_join": False, @@ -166,17 +172,18 @@ class TestCrossFitBookerIsSessionBookable: } } current_time = datetime.now(pytz.timezone("Europe/Paris")) - - result = booker.is_session_bookable(session, current_time) + + result = session_manager.is_session_bookable(session, current_time) assert result is True - + def test_is_session_bookable_booking_window_future(self): """Test session not bookable with booking window in future""" with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) session = { "user_info": { "can_join": False, @@ -185,10 +192,9 @@ class TestCrossFitBookerIsSessionBookable: } } current_time = datetime.now(pytz.timezone("Europe/Paris")) - - result = booker.is_session_bookable(session, current_time) - assert result is False + result = session_manager.is_session_bookable(session, current_time) + assert result is False class TestCrossFitBookerRunBookingCycle: """Test cases for run_booking_cycle method""" @@ -214,7 +220,7 @@ class TestCrossFitBookerRunBookingCycle: # Use current date for the session to ensure it falls within 0-2 day window current_time = datetime.now(pytz.timezone("Europe/Paris")) session_date = current_time.date() - + mock_get_sessions.return_value = { "success": True, "data": { @@ -264,12 +270,12 @@ class TestCrossFitBookerRun: """Test run with booking outside window""" mock_login.return_value = True mock_quit.return_value = None # Prevent actual exit - + # Create a time outside the booking window (19:00) tz = pytz.timezone("Europe/Paris") mock_now = datetime(2025, 7, 25, 19, 0, tzinfo=tz) mock_datetime.now.return_value = mock_now - + # Make sleep return immediately to allow one iteration, then break call_count = 0 def sleep_side_effect(seconds): @@ -279,22 +285,22 @@ class TestCrossFitBookerRun: # Break the loop after first sleep raise KeyboardInterrupt("Test complete") return None - + mock_sleep.side_effect = sleep_side_effect - + booker = CrossFitBooker() - + try: await booker.run() except KeyboardInterrupt: pass # Expected to break the loop - + # Verify login was called mock_login.assert_called_once() - + # Verify run_booking_cycle was NOT called since we're outside the booking window mock_run_booking_cycle.assert_not_called() - + # Verify quit was called (due to KeyboardInterrupt) mock_quit.assert_called_once() @@ -308,25 +314,27 @@ class TestCrossFitBookerQuit: with pytest.raises(SystemExit) as excinfo: booker.quit() assert excinfo.value.code == 0 + class TestCrossFitBookerMatchesPreferredSession: """Test cases for matches_preferred_session method""" - + def test_matches_preferred_session_exact_match(self): """Test exact match with preferred session""" with patch.dict(os.environ, { 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) session = { "start_timestamp": "2025-07-30 18:30:00", "name_activity": "CONDITIONING" } current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris")) - + # Mock PREFERRED_SESSIONS - with patch('crossfit_booker.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]): - result = booker.matches_preferred_session(session, current_time) + with patch('session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]): + result = session_manager.matches_preferred_session(session, current_time) assert result is True def test_matches_preferred_session_fuzzy_match(self): @@ -335,7 +343,8 @@ class TestCrossFitBookerMatchesPreferredSession: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) session = { "start_timestamp": "2025-07-30 18:30:00", "name_activity": "CONDITIONING WORKOUT" @@ -343,8 +352,8 @@ class TestCrossFitBookerMatchesPreferredSession: current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris")) # Mock PREFERRED_SESSIONS - with patch('crossfit_booker.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]): - result = booker.matches_preferred_session(session, current_time) + with patch('session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]): + result = session_manager.matches_preferred_session(session, current_time) assert result is True def test_matches_preferred_session_no_match(self): @@ -353,7 +362,8 @@ class TestCrossFitBookerMatchesPreferredSession: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) session = { "start_timestamp": "2025-07-30 18:30:00", "name_activity": "YOGA" @@ -361,6 +371,6 @@ class TestCrossFitBookerMatchesPreferredSession: current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris")) # Mock PREFERRED_SESSIONS - with patch('crossfit_booker.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]): - result = booker.matches_preferred_session(session, current_time) + with patch('session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]): + result = session_manager.matches_preferred_session(session, current_time) assert result is False \ No newline at end of file