From b6ea2a4ff1cf673776357245728d44162f440954 Mon Sep 17 00:00:00 2001 From: kbe Date: Wed, 1 Oct 2025 17:48:41 +0200 Subject: [PATCH] refactor: Remove CrossfitBooker class and simplify booking system --- AGENT.md => AGENTS.md | 13 +- execute_book_session.py | 43 ++++- main.py | 46 ++++- preferred_sessions.json | 8 +- src/__init__.py | 2 - src/book_crossfit.py | 46 ++++- src/crossfit_booker.py | 193 --------------------- test/test_booking_window.py | 66 -------- test/test_crossfit_booker_auth.py | 46 ++--- test/test_crossfit_booker_final.py | 232 -------------------------- test/test_crossfit_booker_sessions.py | 137 +-------------- tools/execute_book_session.py | 43 ++++- 12 files changed, 207 insertions(+), 668 deletions(-) rename AGENT.md => AGENTS.md (68%) delete mode 100644 src/crossfit_booker.py delete mode 100644 test/test_booking_window.py delete mode 100644 test/test_crossfit_booker_final.py diff --git a/AGENT.md b/AGENTS.md similarity index 68% rename from AGENT.md rename to AGENTS.md index 5ca50e1..7466f9a 100644 --- a/AGENT.md +++ b/AGENTS.md @@ -19,10 +19,17 @@ This file configures the behavior and preferences for the Qwen Code agent for th ### Project Structure Awareness - `main.py`: Entry point for the application. -- `execute_book_session.py`: Core logic for booking sessions. +- `execute_book_session.py`: Alternative execution script for booking sessions. - `preferred_sessions.json`: User preferences for session booking. - `src/`: Directory for source code modules. -- `test_book_session.sh`: Shell script for testing session booking. + - `auth.py`: Authentication handling with bearer token management. + - `booker.py`: Main booking logic with time-based booking windows. + - `session_manager.py`: Session fetching, filtering, and booking operations. + - `session_notifier.py`: Email and Telegram notifications for booking status. + - `session_config.py`: Configuration for preferred session matching. + - `test_book_session.sh`: Shell script for testing session booking. +- `requirements.txt`: Python dependencies. +- `Dockerfile` and `docker-compose.yml`: Container configuration. ### Preferred Tools & Commands - Use `pytest` for running tests. @@ -52,3 +59,5 @@ This file configures the behavior and preferences for the Qwen Code agent for th - When modifying `preferred_sessions.json`, ensure the structure remains valid. - When writing shell scripts, ensure they are executable and follow best practices. - When creating new Python files, ensure they are properly formatted and tested. +- Environment variables required: `CROSSFIT_USERNAME`, `CROSSFIT_PASSWORD`, plus optional email/telegram settings. +- Booking window controlled by `TARGET_RESERVATION_TIME` (default: "20:01") and `BOOKING_WINDOW_END_DELTA_MINUTES` (default: 10). diff --git a/execute_book_session.py b/execute_book_session.py index 30eec64..18807b7 100755 --- a/execute_book_session.py +++ b/execute_book_session.py @@ -4,8 +4,15 @@ Script to demonstrate how to execute the book_session method from crossfit_booke """ import sys +import os import logging -from crossfit_booker import CrossFitBooker +from src.auth import AuthHandler +from src.booker import Booker +from src.session_notifier import SessionNotifier + +# Load environment variables +from dotenv import load_dotenv +load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') @@ -18,12 +25,40 @@ def main(): session_id = sys.argv[1] - # Create an instance of CrossFitBooker - booker = CrossFitBooker() + # Initialize components + auth_handler = AuthHandler( + os.environ.get("CROSSFIT_USERNAME"), + os.environ.get("CROSSFIT_PASSWORD") + ) + + # Initialize notification system (minimal for single session booking) + email_credentials = { + "from": os.environ.get("EMAIL_FROM"), + "to": os.environ.get("EMAIL_TO"), + "password": os.environ.get("EMAIL_PASSWORD") + } + + telegram_credentials = { + "token": os.environ.get("TELEGRAM_TOKEN"), + "chat_id": os.environ.get("TELEGRAM_CHAT_ID") + } + + enable_email = os.environ.get("ENABLE_EMAIL_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + enable_telegram = os.environ.get("ENABLE_TELEGRAM_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + + notifier = SessionNotifier( + email_credentials, + telegram_credentials, + enable_email=enable_email, + enable_telegram=enable_telegram + ) + + # Create an instance of Booker + booker = Booker(auth_handler, notifier) # Login to authenticate print("Attempting to authenticate...") - if not booker.login(): + if not auth_handler.login(): print("Failed to authenticate. Please check your credentials and try again.") sys.exit(1) print("Authentication successful!") diff --git a/main.py b/main.py index f5f523f..94b0208 100755 --- a/main.py +++ b/main.py @@ -5,11 +5,18 @@ This script initializes the CrossFitBooker and starts the booking process. """ import logging -from src.crossfit_booker import CrossFitBooker +import os +from src.auth import AuthHandler +from src.booker import Booker +from src.session_notifier import SessionNotifier + +# Load environment variables +from dotenv import load_dotenv +load_dotenv() def main(): """ - Main function to initialize the CrossFitBooker and start the booking process. + Main function to initialize the Booker and start the booking process. """ # Set up logging logging.basicConfig( @@ -20,11 +27,40 @@ def main(): ] ) - # Initialize the CrossFitBooker - booker = CrossFitBooker() + # Initialize components + auth_handler = AuthHandler( + os.environ.get("CROSSFIT_USERNAME"), + os.environ.get("CROSSFIT_PASSWORD") + ) + + # Initialize notification system + email_credentials = { + "from": os.environ.get("EMAIL_FROM"), + "to": os.environ.get("EMAIL_TO"), + "password": os.environ.get("EMAIL_PASSWORD") + } + + telegram_credentials = { + "token": os.environ.get("TELEGRAM_TOKEN"), + "chat_id": os.environ.get("TELEGRAM_CHAT_ID") + } + + enable_email = os.environ.get("ENABLE_EMAIL_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + enable_telegram = os.environ.get("ENABLE_TELEGRAM_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + + notifier = SessionNotifier( + email_credentials, + telegram_credentials, + enable_email=enable_email, + enable_telegram=enable_telegram + ) + + # Initialize the Booker + booker = Booker(auth_handler, notifier) # Run the booking process - booker.run() + import asyncio + asyncio.run(booker.run()) if __name__ == "__main__": main() \ No newline at end of file diff --git a/preferred_sessions.json b/preferred_sessions.json index 6868785..39c36ad 100644 --- a/preferred_sessions.json +++ b/preferred_sessions.json @@ -1,13 +1,13 @@ [ { - "day_of_week": 2, - "start_time": "18:30", - "session_name_contains": "CONDITIONING" + "day_of_week": 0, + "start_time": "17:45", + "session_name_contains": "WEIGHTLIFTING LOUVRE 3" }, { "day_of_week": 4, "start_time": "17:00", - "session_name_contains": "WEIGHTLIFTING" + "session_name_contains": "CONDITIONING LOUVRE 3" }, { "day_of_week": 5, diff --git a/src/__init__.py b/src/__init__.py index c271611..77d73a9 100644 --- a/src/__init__.py +++ b/src/__init__.py @@ -3,7 +3,6 @@ # Import all public modules to make them available as src.module from .auth import AuthHandler from .booker import Booker -from .crossfit_booker import CrossFitBooker from .session_config import PREFERRED_SESSIONS from .session_manager import SessionManager from .session_notifier import SessionNotifier @@ -11,7 +10,6 @@ from .session_notifier import SessionNotifier __all__ = [ "AuthHandler", "Booker", - "CrossFitBooker", "PREFERRED_SESSIONS", "SessionManager", "SessionNotifier" diff --git a/src/book_crossfit.py b/src/book_crossfit.py index 604dc9c..d4c996b 100755 --- a/src/book_crossfit.py +++ b/src/book_crossfit.py @@ -1,8 +1,15 @@ #!/usr/bin/env python3 import logging +import os import traceback -from crossfit_booker import CrossFitBooker +from src.auth import AuthHandler +from src.booker import Booker +from src.session_notifier import SessionNotifier + +# Load environment variables +from dotenv import load_dotenv +load_dotenv() if __name__ == "__main__": # Configure logging once at script startup @@ -19,16 +26,45 @@ if __name__ == "__main__": logging.getLogger("requests").setLevel(logging.WARNING) logging.info("Logging enhanced with request library noise reduction") - # Create an instance of the CrossFitBooker class - booker = CrossFitBooker() + # Initialize components + auth_handler = AuthHandler( + os.environ.get("CROSSFIT_USERNAME"), + os.environ.get("CROSSFIT_PASSWORD") + ) + + # Initialize notification system + email_credentials = { + "from": os.environ.get("EMAIL_FROM"), + "to": os.environ.get("EMAIL_TO"), + "password": os.environ.get("EMAIL_PASSWORD") + } + + telegram_credentials = { + "token": os.environ.get("TELEGRAM_TOKEN"), + "chat_id": os.environ.get("TELEGRAM_CHAT_ID") + } + + enable_email = os.environ.get("ENABLE_EMAIL_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + enable_telegram = os.environ.get("ENABLE_TELEGRAM_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + + notifier = SessionNotifier( + email_credentials, + telegram_credentials, + enable_email=enable_email, + enable_telegram=enable_telegram + ) + + # Create an instance of the Booker class + booker = Booker(auth_handler, notifier) # Attempt to log in to the CrossFit booking system # TODO: Make a authentication during running request to not get kicked out - if not booker.login(): + if not auth_handler.login(): # If login fails, log the error and exit the script logging.error("Failed to login - Traceback: %s", traceback.format_exc()) exit(1) # Start the continuous booking loop - booker.run() + import asyncio + asyncio.run(booker.run()) logging.info("Script completed") diff --git a/src/crossfit_booker.py b/src/crossfit_booker.py deleted file mode 100644 index 9cb0a44..0000000 --- a/src/crossfit_booker.py +++ /dev/null @@ -1,193 +0,0 @@ -# Native modules -import os -from typing import Dict, Any, Optional, List -from datetime import date, datetime - -# Third-party modules -import requests - -# Import the AuthHandler class -from src.auth import AuthHandler - -# Import the SessionManager class -from src.session_manager import SessionManager - -# Import the Booker class -from src.booker import Booker - -# Import the SessionNotifier class -from src.session_notifier import SessionNotifier - -# Load environment variables -from dotenv import load_dotenv -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") - -class CrossFitBooker: - """ - A simple orchestrator class for the CrossFit booking system. - - This class is responsible for initializing and coordinating the other components - (AuthHandler, SessionManager, and Booker) and provides a unified interface for - interacting with the booking system. - """ - - def __init__(self) -> None: - """ - Initialize the CrossFitBooker with necessary components. - """ - # Initialize the AuthHandler with credentials from environment variables - self.auth_handler = AuthHandler(USERNAME, PASSWORD) - - # Initialize the SessionManager with the AuthHandler - self.session_manager = SessionManager(self.auth_handler) - - # Initialize the SessionNotifier with credentials from environment variables - email_credentials = { - "from": os.environ.get("EMAIL_FROM"), - "to": os.environ.get("EMAIL_TO"), - "password": os.environ.get("EMAIL_PASSWORD") - } - - telegram_credentials = { - "token": os.environ.get("TELEGRAM_TOKEN"), - "chat_id": os.environ.get("TELEGRAM_CHAT_ID") - } - - # Get notification settings from environment variables - enable_email = os.environ.get("ENABLE_EMAIL_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") - enable_telegram = os.environ.get("ENABLE_TELEGRAM_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") - - self.notifier = SessionNotifier( - email_credentials, - telegram_credentials, - enable_email=enable_email, - enable_telegram=enable_telegram - ) - - # Initialize the Booker with the AuthHandler and SessionNotifier - self.booker = Booker(self.auth_handler, self.notifier) - - # Initialize a session for direct API calls - self.session = requests.Session() - - def run(self) -> None: - """ - Start the booking process. - - This method initiates the booking process by running the Booker's main execution loop. - """ - import asyncio - try: - asyncio.run(self.booker.run()) - except Exception as e: - logging.error(f"Error in booking process: {str(e)}") - - 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 login(self) -> bool: - """ - Authenticate and get the bearer token. - - Returns: - bool: True if login is successful, False otherwise. - """ - return self.auth_handler.login() - - 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. - """ - return self.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. - """ - return self.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. - """ - return self.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. - """ - return self.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. - """ - return self.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. - """ - self.session_manager.display_upcoming_sessions(sessions, current_time) - - async def run_booking_cycle(self, current_time: datetime) -> None: - """ - Run one cycle of checking and booking sessions. - - Args: - current_time (datetime): Current time for comparison. - """ - await self.booker.booker(current_time) - - def quit(self) -> None: - """ - Clean up resources and exit the script. - """ - self.booker.quit() diff --git a/test/test_booking_window.py b/test/test_booking_window.py deleted file mode 100644 index 8469047..0000000 --- a/test/test_booking_window.py +++ /dev/null @@ -1,66 +0,0 @@ -#!/usr/bin/env python3 -""" -Test script to verify booking window functionality. -""" -import os -import sys -import logging -from datetime import datetime, timedelta -import pytz - -# Add the parent directory to the path -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) - -from src.crossfit_booker import CrossFitBooker -from src.booker import Booker -from src.auth import AuthHandler - -# Set up logging -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') - -# Mock the login method to avoid actual authentication -def mock_login(self) -> bool: - self.auth_token = "mock_token" - self.user_id = "12345" - return True - -# Test the booking window functionality -def test_booking_window(): - """Test the booking window functionality.""" - # Create a booker instance - booker = CrossFitBooker() - - # Replace the login method with our mock - original_login = AuthHandler.login - AuthHandler.login = mock_login - - # Set up timezone and target time - tz = pytz.timezone("Europe/Paris") - current_time = datetime.now(tz) - - # Get the target time from the environment variable or use default - target_time_str = os.environ.get("TARGET_RESERVATION_TIME", "20:01") - target_hour, target_minute = map(int, target_time_str.split(":")) - target_time = current_time.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0) - - # Calculate booking window end - booking_window_end = target_time + timedelta(minutes=10) - - # Display current time and booking window - logging.info(f"Current time: {current_time}") - logging.info(f"Target booking time: {target_time}") - logging.info(f"Booking window end: {booking_window_end}") - - # Check if we're in the booking window - if target_time <= current_time <= booking_window_end: - logging.info("We are within the booking window!") - else: - logging.info("We are outside the booking window.") - time_diff = (target_time - current_time).total_seconds() - logging.info(f"Next booking window starts in: {time_diff//60} minutes and {time_diff%60:.0f} seconds") - - # Restore the original login method - AuthHandler.login = original_login - -if __name__ == "__main__": - test_booking_window() \ No newline at end of file diff --git a/test/test_crossfit_booker_auth.py b/test/test_crossfit_booker_auth.py index 0eeb9b7..93669a1 100644 --- a/test/test_crossfit_booker_auth.py +++ b/test/test_crossfit_booker_auth.py @@ -12,7 +12,9 @@ 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 src.crossfit_booker import CrossFitBooker +from src.auth import AuthHandler +from src.booker import Booker +from src.session_notifier import SessionNotifier class TestCrossFitBookerAuthHeaders: """Test cases for get_auth_headers method""" @@ -23,8 +25,8 @@ class TestCrossFitBookerAuthHeaders: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - headers = booker.get_auth_headers() + 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" @@ -34,13 +36,13 @@ class TestCrossFitBookerAuthHeaders: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - booker.auth_handler.auth_token = "test_token_123" - headers = booker.get_auth_headers() + 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 TestCrossFitBookerLogin: - """Test cases for login method""" +class TestAuthHandlerLogin: + """Test cases for AuthHandler login method""" @patch('requests.Session.post') def test_login_success(self, mock_post): @@ -69,12 +71,12 @@ class TestCrossFitBookerLogin: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - result = booker.login() + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() assert result is True - assert booker.auth_handler.user_id == "12345" - assert booker.auth_handler.auth_token == "test_bearer_token" + 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): @@ -90,12 +92,12 @@ class TestCrossFitBookerLogin: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - result = booker.login() + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() assert result is False - assert booker.auth_handler.user_id is None - assert booker.auth_handler.auth_token is None + 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): @@ -122,8 +124,8 @@ class TestCrossFitBookerLogin: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - result = booker.login() + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() assert result is False @@ -140,8 +142,8 @@ class TestCrossFitBookerLogin: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - result = booker.login() + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() assert result is False @@ -154,8 +156,8 @@ class TestCrossFitBookerLogin: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - result = booker.login() + auth_handler = AuthHandler('test_user', 'test_pass') + result = auth_handler.login() assert result is False diff --git a/test/test_crossfit_booker_final.py b/test/test_crossfit_booker_final.py deleted file mode 100644 index 6fdc97d..0000000 --- a/test/test_crossfit_booker_final.py +++ /dev/null @@ -1,232 +0,0 @@ -#!/usr/bin/env python3 -""" -Comprehensive unit tests for the CrossFitBooker class in crossfit_booker.py -""" - -import os -import sys -from unittest.mock import Mock, patch -from datetime import date -import requests - -# Add the parent directory to the path to import crossfit_booker -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -from src.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, { - 'CROSSFIT_USERNAME': 'test_user', - 'CROSSFIT_PASSWORD': 'test_pass', - 'EMAIL_FROM': 'from@test.com', - 'EMAIL_TO': 'to@test.com', - 'EMAIL_PASSWORD': 'email_pass', - 'TELEGRAM_TOKEN': 'telegram_token', - 'TELEGRAM_CHAT_ID': '12345' - }): - booker = CrossFitBooker() - 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): - try: - 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, { - 'CROSSFIT_USERNAME': 'test_user' - # Missing PASSWORD - }, clear=True): - try: - CrossFitBooker() - 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, { - 'CROSSFIT_USERNAME': 'test_user', - 'CROSSFIT_PASSWORD': 'test_pass' - }): - booker = CrossFitBooker() - 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, { - 'CROSSFIT_USERNAME': 'test_user', - 'CROSSFIT_PASSWORD': 'test_pass' - }): - booker = CrossFitBooker() - 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""" - # 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] - - with patch.dict(os.environ, { - 'CROSSFIT_USERNAME': 'test_user', - 'CROSSFIT_PASSWORD': 'test_pass' - }): - booker = CrossFitBooker() - result = booker.login() - - assert result is True - 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""" - mock_response = Mock() - 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.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 - -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)) - assert result is None - - @patch('requests.Session.post') - def test_get_available_sessions_success(self, mock_post): - """Test successful get_available_sessions""" - mock_response = Mock() - mock_response.status_code = 200 - mock_response.json.return_value = { - "success": True, - "data": { - "activities_calendar": [ - {"id": "1", "name": "Test Session"} - ] - } - } - - mock_post.return_value = mock_response - - with patch.dict(os.environ, { - 'CROSSFIT_USERNAME': 'test_user', - 'CROSSFIT_PASSWORD': 'test_pass' - }): - booker = CrossFitBooker() - 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_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 452b4ae..97a5609 100644 --- a/test/test_crossfit_booker_sessions.py +++ b/test/test_crossfit_booker_sessions.py @@ -13,9 +13,10 @@ import pytz # Add the parent directory to the path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from src.crossfit_booker import CrossFitBooker -from src.session_manager import SessionManager from src.auth import AuthHandler +from src.booker import Booker +from src.session_notifier import SessionNotifier +from src.session_manager import SessionManager @@ -75,11 +76,12 @@ class TestCrossFitBookerGetAvailableSessions: 'CROSSFIT_USERNAME': 'test_user', 'CROSSFIT_PASSWORD': 'test_pass' }): - booker = CrossFitBooker() - booker.auth_handler.auth_token = "test_token" - booker.auth_handler.user_id = "12345" + auth_handler = AuthHandler('test_user', 'test_pass') + session_manager = SessionManager(auth_handler) + auth_handler.auth_token = "test_token" + auth_handler.user_id = "12345" - result = booker.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) + result = session_manager.get_available_sessions(date(2025, 7, 24), date(2025, 7, 25)) assert result is None @@ -198,129 +200,6 @@ class TestCrossFitBookerIsSessionBookable: result = session_manager.is_session_bookable(session, current_time) assert result is False -class TestCrossFitBookerExcuteCycle: - """Test cases for execute_cycle method""" - - @patch('src.crossfit_booker.CrossFitBooker.get_available_sessions') - @patch('src.crossfit_booker.CrossFitBooker.is_session_bookable') - @patch('src.crossfit_booker.CrossFitBooker.matches_preferred_session') - @patch('src.crossfit_booker.CrossFitBooker.book_session') - async def test_run_booking_cycle_no_sessions(self, mock_book_session, mock_matches_preferred, mock_is_bookable, mock_get_sessions): - """Test run_booking_cycle with no available sessions""" - mock_get_sessions.return_value = {"success": False} - booker = CrossFitBooker() - # Mock the auth_token and user_id to avoid authentication errors - booker.auth_handler.auth_token = "test_token" - booker.auth_handler.user_id = "12345" - # Mock the booker method to use our mocked methods - with patch.object(booker.booker, 'get_available_sessions', mock_get_sessions): - await booker.run_booking_cycle(datetime.now(pytz.timezone("Europe/Paris"))) - mock_get_sessions.assert_called_once() - mock_book_session.assert_not_called() - - @patch('src.crossfit_booker.CrossFitBooker.get_available_sessions') - @patch('src.crossfit_booker.CrossFitBooker.is_session_bookable') - @patch('src.crossfit_booker.CrossFitBooker.matches_preferred_session') - @patch('src.crossfit_booker.CrossFitBooker.book_session') - async def test_run_booking_cycle_with_sessions(self, mock_book_session, mock_matches_preferred, mock_is_bookable, mock_get_sessions): - """Test run_booking_cycle with available sessions""" - # 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": { - "activities_calendar": [ - { - "id_activity_calendar": "1", - "name_activity": "CONDITIONING", - "start_timestamp": session_date.strftime("%Y-%m-%d") + " 18:30:00", - "user_info": {"can_join": True} - } - ] - } - } - mock_is_bookable.return_value = True - mock_matches_preferred.return_value = True - mock_book_session.return_value = True - - booker = CrossFitBooker() - # Mock the auth_token and user_id to avoid authentication errors - booker.auth_handler.auth_token = "test_token" - booker.auth_handler.user_id = "12345" - # Mock the booker method to use our mocked methods - with patch.object(booker.booker, 'get_available_sessions', mock_get_sessions): - with patch.object(booker.booker, 'is_session_bookable', mock_is_bookable): - with patch.object(booker.booker, 'matches_preferred_session', mock_matches_preferred): - with patch.object(booker.booker, 'book_session', mock_book_session): - await booker.run_booking_cycle(current_time) - - mock_get_sessions.assert_called_once() - mock_is_bookable.assert_called_once() - mock_matches_preferred.assert_called_once() - mock_book_session.assert_called_once() - assert mock_book_session.call_count == 1 - -class TestCrossFitBookerRun: - """Test cases for run method""" - - def test_run_auth_failure(self): - """Test run with authentication failure""" - with patch('src.crossfit_booker.CrossFitBooker.login', return_value=False) as mock_login: - booker = CrossFitBooker() - # Test the authentication failure path through the booker - result = booker.login() - assert result is False - mock_login.assert_called_once() - - def test_run_booking_outside_window(self): - """Test run with booking outside window""" - with patch('src.booker.Booker.run') as mock_run: - with patch('datetime.datetime') as mock_datetime: - with patch('time.sleep') as mock_sleep: - # 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): - nonlocal call_count - call_count += 1 - if call_count >= 1: - # Break the loop after first sleep - raise KeyboardInterrupt("Test complete") - return None - - mock_sleep.side_effect = sleep_side_effect - - booker = CrossFitBooker() - - # Test the booking window logic directly - target_hour, target_minute = map(int, "20:01".split(":")) - target_time = datetime.now(tz).replace(hour=target_hour, minute=target_minute, second=0, microsecond=0) - booking_window_end = target_time + timedelta(minutes=10) - - # Current time is outside the booking window - assert not (target_time <= mock_now <= booking_window_end) - - # Run the booker to trigger the login - booker.run() - - # Verify run was called - mock_run.assert_called_once() - -class TestCrossFitBookerQuit: - """Test cases for quit method""" - - def test_quit(self): - """Test quit method""" - booker = CrossFitBooker() - with pytest.raises(SystemExit) as excinfo: - booker.quit() - assert excinfo.value.code == 0 class TestCrossFitBookerMatchesPreferredSession: """Test cases for matches_preferred_session method""" diff --git a/tools/execute_book_session.py b/tools/execute_book_session.py index 30eec64..18807b7 100644 --- a/tools/execute_book_session.py +++ b/tools/execute_book_session.py @@ -4,8 +4,15 @@ Script to demonstrate how to execute the book_session method from crossfit_booke """ import sys +import os import logging -from crossfit_booker import CrossFitBooker +from src.auth import AuthHandler +from src.booker import Booker +from src.session_notifier import SessionNotifier + +# Load environment variables +from dotenv import load_dotenv +load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') @@ -18,12 +25,40 @@ def main(): session_id = sys.argv[1] - # Create an instance of CrossFitBooker - booker = CrossFitBooker() + # Initialize components + auth_handler = AuthHandler( + os.environ.get("CROSSFIT_USERNAME"), + os.environ.get("CROSSFIT_PASSWORD") + ) + + # Initialize notification system (minimal for single session booking) + email_credentials = { + "from": os.environ.get("EMAIL_FROM"), + "to": os.environ.get("EMAIL_TO"), + "password": os.environ.get("EMAIL_PASSWORD") + } + + telegram_credentials = { + "token": os.environ.get("TELEGRAM_TOKEN"), + "chat_id": os.environ.get("TELEGRAM_CHAT_ID") + } + + enable_email = os.environ.get("ENABLE_EMAIL_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + enable_telegram = os.environ.get("ENABLE_TELEGRAM_NOTIFICATIONS", "true").lower() in ("true", "1", "yes") + + notifier = SessionNotifier( + email_credentials, + telegram_credentials, + enable_email=enable_email, + enable_telegram=enable_telegram + ) + + # Create an instance of Booker + booker = Booker(auth_handler, notifier) # Login to authenticate print("Attempting to authenticate...") - if not booker.login(): + if not auth_handler.login(): print("Failed to authenticate. Please check your credentials and try again.") sys.exit(1) print("Authentication successful!")