kbe/crossfit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor code and split into multiple files #9

Merged
kbe merged 6 commits from develop into main 2025-08-11 23:44:32 +00:00
Conversation 0 Commits 6 Files Changed 27 +1079 -646
11 changed files with 1079 additions and 646 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 90230832ee - Show all commits

116
auth.py Normal file
View File

@@ -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

2
book_crossfit.py
View File

@@ -31,5 +31,5 @@ if __name__ == "__main__":
exit(1)
# Start the continuous booking loop
asyncio.run(booker.run())
booker.run()
logging.info("Script completed")

310
booker.py Normal file
View File

@@ -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)

517
crossfit_booker.py
View File

@@ -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())

31
main.py Executable file
View File

@@ -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()

295
session_manager.py Normal file
View File

@@ -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}")

135
test/test_auth.py Normal file
View File

@@ -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"])

3
test/test_crossfit_booker.py
View File

@@ -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

62
test/test_crossfit_booker_auth.py
View File

@@ -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"])

90
test/test_crossfit_booker_final.py
View File

@@ -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

164
test/test_crossfit_booker_sessions.py
View File

@@ -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