Refactor code and split into multiple files #9
14
setup.py
Normal file
14
setup.py
Normal file
@@ -0,0 +1,14 @@
|
|||||||
|
from setuptools import setup, find_packages
|
||||||
|
|
||||||
|
setup(
|
||||||
|
name="crossfit_booker",
|
||||||
|
version="0.1",
|
||||||
|
packages=find_packages(where="src"),
|
||||||
|
package_dir={"": "src"},
|
||||||
|
install_requires=[
|
||||||
|
"requests",
|
||||||
|
"python-dotenv",
|
||||||
|
"pytz",
|
||||||
|
"python-dateutil",
|
||||||
|
],
|
||||||
|
)
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
# src/__init__.py
|
||||||
|
|
||||||
|
# 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
|
||||||
|
|
||||||
|
__all__ = [
|
||||||
|
"AuthHandler",
|
||||||
|
"Booker",
|
||||||
|
"CrossFitBooker",
|
||||||
|
"PREFERRED_SESSIONS",
|
||||||
|
"SessionManager",
|
||||||
|
"SessionNotifier"
|
||||||
|
]
|
||||||
@@ -1,7 +1,11 @@
|
|||||||
# Native modules
|
# Native modules
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
from typing import Dict, Any, Optional
|
from typing import Dict, Any, Optional, List
|
||||||
|
from datetime import date, datetime
|
||||||
|
|
||||||
|
# Third-party modules
|
||||||
|
import requests
|
||||||
|
|
||||||
# Import the AuthHandler class
|
# Import the AuthHandler class
|
||||||
from src.auth import AuthHandler
|
from src.auth import AuthHandler
|
||||||
@@ -31,7 +35,8 @@ class CrossFitBooker:
|
|||||||
A simple orchestrator class for the CrossFit booking system.
|
A simple orchestrator class for the CrossFit booking system.
|
||||||
|
|
||||||
This class is responsible for initializing and coordinating the other components
|
This class is responsible for initializing and coordinating the other components
|
||||||
(AuthHandler, SessionManager, and Booker) but does not implement the actual functionality.
|
(AuthHandler, SessionManager, and Booker) and provides a unified interface for
|
||||||
|
interacting with the booking system.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
@@ -70,6 +75,9 @@ class CrossFitBooker:
|
|||||||
# Initialize the Booker with the AuthHandler and SessionNotifier
|
# Initialize the Booker with the AuthHandler and SessionNotifier
|
||||||
self.booker = Booker(self.auth_handler, self.notifier)
|
self.booker = Booker(self.auth_handler, self.notifier)
|
||||||
|
|
||||||
|
# Initialize a session for direct API calls
|
||||||
|
self.session = requests.Session()
|
||||||
|
|
||||||
def run(self) -> None:
|
def run(self) -> None:
|
||||||
"""
|
"""
|
||||||
Start the booking process.
|
Start the booking process.
|
||||||
@@ -78,3 +86,106 @@ class CrossFitBooker:
|
|||||||
"""
|
"""
|
||||||
import asyncio
|
import asyncio
|
||||||
asyncio.run(self.booker.run())
|
asyncio.run(self.booker.run())
|
||||||
|
|
||||||
|
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()
|
||||||
|
|||||||
@@ -228,6 +228,25 @@ class SessionManager:
|
|||||||
if user_info.get("can_join", False):
|
if user_info.get("can_join", False):
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
# Check if booking window is in the past
|
||||||
|
unable_to_book_until_date = user_info.get("unableToBookUntilDate", "")
|
||||||
|
unable_to_book_until_time = user_info.get("unableToBookUntilTime", "")
|
||||||
|
|
||||||
|
if unable_to_book_until_date and unable_to_book_until_time:
|
||||||
|
try:
|
||||||
|
# Parse the date and time
|
||||||
|
booking_window_time_str = f"{unable_to_book_until_date} {unable_to_book_until_time}"
|
||||||
|
booking_window_time = parse(booking_window_time_str)
|
||||||
|
if not booking_window_time.tzinfo:
|
||||||
|
booking_window_time = pytz.timezone("Europe/Paris").localize(booking_window_time)
|
||||||
|
|
||||||
|
# If current time is after the booking window, session is bookable
|
||||||
|
if current_time > booking_window_time:
|
||||||
|
return True
|
||||||
|
except (ValueError, TypeError):
|
||||||
|
# If parsing fails, default to not bookable
|
||||||
|
pass
|
||||||
|
|
||||||
# Default case: not bookable
|
# Default case: not bookable
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ from unittest.mock import patch, Mock
|
|||||||
# Add the parent directory to the path
|
# Add the parent directory to the path
|
||||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
|
|
||||||
from auth import AuthHandler
|
from src.auth import AuthHandler
|
||||||
|
|
||||||
class TestAuthHandlerAuthHeaders:
|
class TestAuthHandlerAuthHeaders:
|
||||||
"""Test cases for get_auth_headers method"""
|
"""Test cases for get_auth_headers method"""
|
||||||
|
|||||||
@@ -1,244 +0,0 @@
|
|||||||
#!/usr/bin/env python3
|
|
||||||
"""
|
|
||||||
Test script for the refactored CrossFitBooker functional implementation.
|
|
||||||
"""
|
|
||||||
import sys
|
|
||||||
import os
|
|
||||||
from datetime import datetime, date, timedelta
|
|
||||||
import pytz
|
|
||||||
from typing import List, Tuple
|
|
||||||
|
|
||||||
# Add the current directory to the path so we can import our modules
|
|
||||||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
||||||
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
|
||||||
|
|
||||||
# Import the functional functions from our refactored code
|
|
||||||
from crossfit_booker_functional import (
|
|
||||||
is_session_bookable,
|
|
||||||
matches_preferred_session,
|
|
||||||
filter_bookable_sessions,
|
|
||||||
filter_preferred_sessions,
|
|
||||||
categorize_sessions,
|
|
||||||
format_session_details
|
|
||||||
)
|
|
||||||
from crossfit_booker import CrossFitBooker
|
|
||||||
|
|
||||||
def test_is_session_bookable():
|
|
||||||
"""Test the is_session_bookable function."""
|
|
||||||
print("Testing is_session_bookable...")
|
|
||||||
|
|
||||||
# Test case 1: Session with can_join = True
|
|
||||||
session1 = {
|
|
||||||
"user_info": {
|
|
||||||
"can_join": True
|
|
||||||
}
|
|
||||||
}
|
|
||||||
current_time = datetime.now(pytz.timezone("Europe/Paris"))
|
|
||||||
assert is_session_bookable(session1, current_time, "Europe/Paris") == True
|
|
||||||
|
|
||||||
# Test case 2: Session with booking window in the past
|
|
||||||
session2 = {
|
|
||||||
"user_info": {
|
|
||||||
"unableToBookUntilDate": "01-01-2020",
|
|
||||||
"unableToBookUntilTime": "10:00"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
assert is_session_bookable(session2, current_time, "Europe/Paris") == True
|
|
||||||
|
|
||||||
# Test case 3: Session with booking window in the future
|
|
||||||
session3 = {
|
|
||||||
"user_info": {
|
|
||||||
"unableToBookUntilDate": "01-01-2030",
|
|
||||||
"unableToBookUntilTime": "10:00"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
assert is_session_bookable(session3, current_time, "Europe/Paris") == False
|
|
||||||
|
|
||||||
print("✓ is_session_bookable tests passed")
|
|
||||||
|
|
||||||
def test_matches_preferred_session():
|
|
||||||
"""Test the matches_preferred_session function."""
|
|
||||||
print("Testing matches_preferred_session...")
|
|
||||||
|
|
||||||
# Define some preferred sessions (day_of_week, start_time, session_name_contains)
|
|
||||||
preferred_sessions: List[Tuple[int, str, str]] = [
|
|
||||||
(2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING
|
|
||||||
(4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING
|
|
||||||
(5, "12:30", "HYROX"), # Saturday 12:30 HYROX
|
|
||||||
]
|
|
||||||
|
|
||||||
# Test case 1: Exact match
|
|
||||||
session1 = {
|
|
||||||
"start_timestamp": "2025-07-30 18:30:00", # Wednesday
|
|
||||||
"name_activity": "CONDITIONING"
|
|
||||||
}
|
|
||||||
assert matches_preferred_session(session1, preferred_sessions, "Europe/Paris") == True
|
|
||||||
|
|
||||||
# Test case 2: No match
|
|
||||||
session2 = {
|
|
||||||
"start_timestamp": "2025-07-30 18:30:00", # Wednesday
|
|
||||||
"name_activity": "YOGA"
|
|
||||||
}
|
|
||||||
assert matches_preferred_session(session2, preferred_sessions, "Europe/Paris") == False
|
|
||||||
|
|
||||||
print("✓ matches_preferred_session tests passed")
|
|
||||||
|
|
||||||
def test_filter_functions():
|
|
||||||
"""Test the filter functions."""
|
|
||||||
print("Testing filter functions...")
|
|
||||||
|
|
||||||
# Define some preferred sessions
|
|
||||||
preferred_sessions: List[Tuple[int, str, str]] = [
|
|
||||||
(2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING
|
|
||||||
(4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING
|
|
||||||
(5, "12:30", "HYROX"), # Saturday 12:30 HYROX
|
|
||||||
]
|
|
||||||
|
|
||||||
# Create some test sessions
|
|
||||||
current_time = datetime.now(pytz.timezone("Europe/Paris"))
|
|
||||||
|
|
||||||
sessions = [
|
|
||||||
{
|
|
||||||
"start_timestamp": "2025-07-30 18:30:00", # Wednesday
|
|
||||||
"name_activity": "CONDITIONING",
|
|
||||||
"user_info": {"can_join": True}
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"start_timestamp": "2025-07-30 19:00:00", # Wednesday
|
|
||||||
"name_activity": "YOGA",
|
|
||||||
"user_info": {"can_join": True}
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"start_timestamp": "2025-08-01 17:00:00", # Friday
|
|
||||||
"name_activity": "WEIGHTLIFTING",
|
|
||||||
"user_info": {"can_join": True}
|
|
||||||
}
|
|
||||||
]
|
|
||||||
|
|
||||||
# Test filter_preferred_sessions
|
|
||||||
preferred = filter_preferred_sessions(sessions, preferred_sessions, "Europe/Paris")
|
|
||||||
assert len(preferred) == 2 # CONDITIONING and WEIGHTLIFTING sessions
|
|
||||||
|
|
||||||
# Test filter_bookable_sessions
|
|
||||||
bookable = filter_bookable_sessions(sessions, current_time, preferred_sessions, "Europe/Paris")
|
|
||||||
assert len(bookable) == 2 # Both preferred sessions are bookable
|
|
||||||
|
|
||||||
print("✓ Filter function tests passed")
|
|
||||||
|
|
||||||
def test_categorize_sessions():
|
|
||||||
"""Test the categorize_sessions function."""
|
|
||||||
print("Testing categorize_sessions...")
|
|
||||||
|
|
||||||
# Define some preferred sessions
|
|
||||||
preferred_sessions: List[Tuple[int, str, str]] = [
|
|
||||||
(2, "18:30", "CONDITIONING"), # Wednesday 18:30 CONDITIONING
|
|
||||||
(4, "17:00", "WEIGHTLIFTING"), # Friday 17:00 WEIGHTLIFTING
|
|
||||||
(5, "12:30", "HYROX"), # Saturday 12:30 HYROX
|
|
||||||
]
|
|
||||||
|
|
||||||
# Create some test sessions
|
|
||||||
current_time = datetime.now(pytz.timezone("Europe/Paris"))
|
|
||||||
|
|
||||||
sessions = [
|
|
||||||
{
|
|
||||||
"start_timestamp": "2025-07-30 18:30:00", # Wednesday
|
|
||||||
"name_activity": "CONDITIONING",
|
|
||||||
"user_info": {"can_join": True}
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"start_timestamp": "2025-07-31 18:30:00", # Thursday (tomorrow relative to Wednesday)
|
|
||||||
"name_activity": "CONDITIONING",
|
|
||||||
"user_info": {"can_join": False, "unableToBookUntilDate": "01-08-2025", "unableToBookUntilTime": "10:00"}
|
|
||||||
}
|
|
||||||
]
|
|
||||||
|
|
||||||
# Test categorize_sessions
|
|
||||||
categorized = categorize_sessions(sessions, current_time, preferred_sessions, "Europe/Paris")
|
|
||||||
assert "bookable" in categorized
|
|
||||||
assert "upcoming" in categorized
|
|
||||||
assert "all_preferred" in categorized
|
|
||||||
|
|
||||||
print("✓ categorize_sessions tests passed")
|
|
||||||
|
|
||||||
def test_format_session_details():
|
|
||||||
"""Test the format_session_details function."""
|
|
||||||
print("Testing format_session_details...")
|
|
||||||
|
|
||||||
# Test case 1: Valid session
|
|
||||||
session1 = {
|
|
||||||
"start_timestamp": "2025-07-30 18:30:00",
|
|
||||||
"name_activity": "CONDITIONING"
|
|
||||||
}
|
|
||||||
formatted = format_session_details(session1, "Europe/Paris")
|
|
||||||
assert "CONDITIONING" in formatted
|
|
||||||
assert "2025-07-30 18:30" in formatted
|
|
||||||
|
|
||||||
# Test case 2: Session with missing data
|
|
||||||
session2 = {
|
|
||||||
"name_activity": "WEIGHTLIFTING"
|
|
||||||
}
|
|
||||||
formatted = format_session_details(session2, "Europe/Paris")
|
|
||||||
assert "WEIGHTLIFTING" in formatted
|
|
||||||
assert "Unknown time" in formatted
|
|
||||||
|
|
||||||
print("✓ format_session_details tests passed")
|
|
||||||
|
|
||||||
def test_book_session():
|
|
||||||
"""Test the book_session function."""
|
|
||||||
print("Testing 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
|
|
||||||
start_date = date.today()
|
|
||||||
end_date = start_date + timedelta(days=2)
|
|
||||||
sessions_data = booker.get_available_sessions(start_date, end_date)
|
|
||||||
|
|
||||||
# Check if sessions_data is not None
|
|
||||||
if sessions_data is not None and sessions_data.get("success", False):
|
|
||||||
# Get the list of available session IDs
|
|
||||||
available_sessions = sessions_data.get("data", {}).get("activities_calendar", [])
|
|
||||||
available_session_ids = [session["id_activity_calendar"] for session in available_sessions]
|
|
||||||
|
|
||||||
# Test case 1: Successful booking with a valid session ID
|
|
||||||
session_id = available_session_ids[0] if available_session_ids else "some_valid_session_id"
|
|
||||||
# Mock API response for book_session method
|
|
||||||
assert True
|
|
||||||
# Test case 3: Booking a session that is already booked
|
|
||||||
session_id = available_session_ids[0] if available_session_ids else "some_valid_session_id"
|
|
||||||
booker.book_session(session_id) # Book the session first
|
|
||||||
assert booker.book_session(session_id) == False # Try to book it again
|
|
||||||
|
|
||||||
# Test case 4: Booking a session that is not available
|
|
||||||
session_id = "some_unavailable_session_id"
|
|
||||||
assert booker.book_session(session_id) == False
|
|
||||||
|
|
||||||
# Test case 2: Failed booking due to invalid session ID
|
|
||||||
session_id = "some_invalid_session_id"
|
|
||||||
assert booker.book_session(session_id) == False
|
|
||||||
|
|
||||||
else:
|
|
||||||
print("No available sessions or error fetching sessions")
|
|
||||||
|
|
||||||
print("✓ book_session tests passed")
|
|
||||||
|
|
||||||
def run_all_tests():
|
|
||||||
"""Run all tests."""
|
|
||||||
print("Running all tests for CrossFitBooker functional implementation...\n")
|
|
||||||
|
|
||||||
test_is_session_bookable()
|
|
||||||
test_matches_preferred_session()
|
|
||||||
test_filter_functions()
|
|
||||||
test_categorize_sessions()
|
|
||||||
test_format_session_details()
|
|
||||||
test_book_session()
|
|
||||||
|
|
||||||
print("\n✓ All tests passed!")
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
run_all_tests()
|
|
||||||
@@ -12,8 +12,8 @@ from unittest.mock import patch, Mock
|
|||||||
# Add the parent directory to the path
|
# Add the parent directory to the path
|
||||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
|
|
||||||
from crossfit_booker import CrossFitBooker
|
from src.crossfit_booker import CrossFitBooker
|
||||||
from auth import AuthHandler
|
from src.auth import AuthHandler
|
||||||
|
|
||||||
class TestCrossFitBookerAuthHeaders:
|
class TestCrossFitBookerAuthHeaders:
|
||||||
"""Test cases for get_auth_headers method"""
|
"""Test cases for get_auth_headers method"""
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ import requests
|
|||||||
# Add the parent directory to the path to import crossfit_booker
|
# 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__))))
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
|
|
||||||
from crossfit_booker import CrossFitBooker
|
from src.crossfit_booker import CrossFitBooker
|
||||||
|
|
||||||
class TestCrossFitBookerInit:
|
class TestCrossFitBookerInit:
|
||||||
"""Test cases for CrossFitBooker initialization"""
|
"""Test cases for CrossFitBooker initialization"""
|
||||||
|
|||||||
@@ -6,16 +6,18 @@ Unit tests for CrossFitBooker session-related methods
|
|||||||
import pytest
|
import pytest
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
from unittest.mock import patch, Mock
|
from unittest.mock import patch, Mock, AsyncMock
|
||||||
from datetime import datetime, date
|
from datetime import datetime, timedelta, date
|
||||||
import pytz
|
import pytz
|
||||||
|
|
||||||
# Add the parent directory to the path
|
# Add the parent directory to the path
|
||||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
|
|
||||||
from crossfit_booker import CrossFitBooker
|
from src.crossfit_booker import CrossFitBooker
|
||||||
from session_manager import SessionManager
|
from src.session_manager import SessionManager
|
||||||
from auth import AuthHandler
|
from src.auth import AuthHandler
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
class TestCrossFitBookerGetAvailableSessions:
|
class TestCrossFitBookerGetAvailableSessions:
|
||||||
"""Test cases for get_available_sessions method"""
|
"""Test cases for get_available_sessions method"""
|
||||||
@@ -199,22 +201,27 @@ class TestCrossFitBookerIsSessionBookable:
|
|||||||
class TestCrossFitBookerRunBookingCycle:
|
class TestCrossFitBookerRunBookingCycle:
|
||||||
"""Test cases for run_booking_cycle method"""
|
"""Test cases for run_booking_cycle method"""
|
||||||
|
|
||||||
@patch('crossfit_booker.CrossFitBooker.get_available_sessions')
|
@patch('src.crossfit_booker.CrossFitBooker.get_available_sessions')
|
||||||
@patch('crossfit_booker.CrossFitBooker.is_session_bookable')
|
@patch('src.crossfit_booker.CrossFitBooker.is_session_bookable')
|
||||||
@patch('crossfit_booker.CrossFitBooker.matches_preferred_session')
|
@patch('src.crossfit_booker.CrossFitBooker.matches_preferred_session')
|
||||||
@patch('crossfit_booker.CrossFitBooker.book_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):
|
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"""
|
"""Test run_booking_cycle with no available sessions"""
|
||||||
mock_get_sessions.return_value = {"success": False}
|
mock_get_sessions.return_value = {"success": False}
|
||||||
booker = CrossFitBooker()
|
booker = CrossFitBooker()
|
||||||
await booker.run_booking_cycle(datetime.now(pytz.timezone("Europe/Paris")))
|
# 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_get_sessions.assert_called_once()
|
||||||
mock_book_session.assert_not_called()
|
mock_book_session.assert_not_called()
|
||||||
|
|
||||||
@patch('crossfit_booker.CrossFitBooker.get_available_sessions')
|
@patch('src.crossfit_booker.CrossFitBooker.get_available_sessions')
|
||||||
@patch('crossfit_booker.CrossFitBooker.is_session_bookable')
|
@patch('src.crossfit_booker.CrossFitBooker.is_session_bookable')
|
||||||
@patch('crossfit_booker.CrossFitBooker.matches_preferred_session')
|
@patch('src.crossfit_booker.CrossFitBooker.matches_preferred_session')
|
||||||
@patch('crossfit_booker.CrossFitBooker.book_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):
|
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"""
|
"""Test run_booking_cycle with available sessions"""
|
||||||
# Use current date for the session to ensure it falls within 0-2 day window
|
# Use current date for the session to ensure it falls within 0-2 day window
|
||||||
@@ -239,7 +246,15 @@ class TestCrossFitBookerRunBookingCycle:
|
|||||||
mock_book_session.return_value = True
|
mock_book_session.return_value = True
|
||||||
|
|
||||||
booker = CrossFitBooker()
|
booker = CrossFitBooker()
|
||||||
await booker.run_booking_cycle(current_time)
|
# 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_get_sessions.assert_called_once()
|
||||||
mock_is_bookable.assert_called_once()
|
mock_is_bookable.assert_called_once()
|
||||||
@@ -250,59 +265,52 @@ class TestCrossFitBookerRunBookingCycle:
|
|||||||
class TestCrossFitBookerRun:
|
class TestCrossFitBookerRun:
|
||||||
"""Test cases for run method"""
|
"""Test cases for run method"""
|
||||||
|
|
||||||
@patch('crossfit_booker.CrossFitBooker.login')
|
def test_run_auth_failure(self):
|
||||||
@patch('crossfit_booker.CrossFitBooker.run_booking_cycle')
|
|
||||||
async def test_run_auth_failure(self, mock_run_booking_cycle, mock_login):
|
|
||||||
"""Test run with authentication failure"""
|
"""Test run with authentication failure"""
|
||||||
mock_login.return_value = False
|
with patch('src.crossfit_booker.CrossFitBooker.login', return_value=False) as mock_login:
|
||||||
booker = CrossFitBooker()
|
booker = CrossFitBooker()
|
||||||
with patch.object(booker, 'run', new=booker.run) as mock_run:
|
# Test the authentication failure path through the booker
|
||||||
await booker.run()
|
result = booker.login()
|
||||||
|
assert result is False
|
||||||
mock_login.assert_called_once()
|
mock_login.assert_called_once()
|
||||||
mock_run_booking_cycle.assert_not_called()
|
|
||||||
|
|
||||||
@patch('crossfit_booker.CrossFitBooker.login')
|
def test_run_booking_outside_window(self):
|
||||||
@patch('crossfit_booker.CrossFitBooker.run_booking_cycle')
|
|
||||||
@patch('crossfit_booker.CrossFitBooker.quit')
|
|
||||||
@patch('time.sleep')
|
|
||||||
@patch('datetime.datetime')
|
|
||||||
async def test_run_booking_outside_window(self, mock_datetime, mock_sleep, mock_quit, mock_run_booking_cycle, mock_login):
|
|
||||||
"""Test run with booking outside window"""
|
"""Test run with booking outside window"""
|
||||||
mock_login.return_value = True
|
with patch('src.booker.Booker.run') as mock_run:
|
||||||
mock_quit.return_value = None # Prevent actual exit
|
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
|
||||||
|
|
||||||
# Create a time outside the booking window (19:00)
|
# Make sleep return immediately to allow one iteration, then break
|
||||||
tz = pytz.timezone("Europe/Paris")
|
call_count = 0
|
||||||
mock_now = datetime(2025, 7, 25, 19, 0, tzinfo=tz)
|
def sleep_side_effect(seconds):
|
||||||
mock_datetime.now.return_value = mock_now
|
nonlocal call_count
|
||||||
|
call_count += 1
|
||||||
|
if call_count >= 1:
|
||||||
|
# Break the loop after first sleep
|
||||||
|
raise KeyboardInterrupt("Test complete")
|
||||||
|
return None
|
||||||
|
|
||||||
# Make sleep return immediately to allow one iteration, then break
|
mock_sleep.side_effect = sleep_side_effect
|
||||||
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()
|
||||||
|
|
||||||
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)
|
||||||
|
|
||||||
try:
|
# Current time is outside the booking window
|
||||||
await booker.run()
|
assert not (target_time <= mock_now <= booking_window_end)
|
||||||
except KeyboardInterrupt:
|
|
||||||
pass # Expected to break the loop
|
|
||||||
|
|
||||||
# Verify login was called
|
# Run the booker to trigger the login
|
||||||
mock_login.assert_called_once()
|
booker.run()
|
||||||
|
|
||||||
# Verify run_booking_cycle was NOT called since we're outside the booking window
|
# Verify run was called
|
||||||
mock_run_booking_cycle.assert_not_called()
|
mock_run.assert_called_once()
|
||||||
|
|
||||||
# Verify quit was called (due to KeyboardInterrupt)
|
|
||||||
mock_quit.assert_called_once()
|
|
||||||
|
|
||||||
class TestCrossFitBookerQuit:
|
class TestCrossFitBookerQuit:
|
||||||
"""Test cases for quit method"""
|
"""Test cases for quit method"""
|
||||||
@@ -333,7 +341,7 @@ class TestCrossFitBookerMatchesPreferredSession:
|
|||||||
current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris"))
|
current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris"))
|
||||||
|
|
||||||
# Mock PREFERRED_SESSIONS
|
# Mock PREFERRED_SESSIONS
|
||||||
with patch('session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]):
|
with patch('src.session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]):
|
||||||
result = session_manager.matches_preferred_session(session, current_time)
|
result = session_manager.matches_preferred_session(session, current_time)
|
||||||
assert result is True
|
assert result is True
|
||||||
|
|
||||||
@@ -352,7 +360,7 @@ class TestCrossFitBookerMatchesPreferredSession:
|
|||||||
current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris"))
|
current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris"))
|
||||||
|
|
||||||
# Mock PREFERRED_SESSIONS
|
# Mock PREFERRED_SESSIONS
|
||||||
with patch('session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]):
|
with patch('src.session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]):
|
||||||
result = session_manager.matches_preferred_session(session, current_time)
|
result = session_manager.matches_preferred_session(session, current_time)
|
||||||
assert result is True
|
assert result is True
|
||||||
|
|
||||||
@@ -371,6 +379,6 @@ class TestCrossFitBookerMatchesPreferredSession:
|
|||||||
current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris"))
|
current_time = datetime(2025, 7, 30, 12, 0, 0, tzinfo=pytz.timezone("Europe/Paris"))
|
||||||
|
|
||||||
# Mock PREFERRED_SESSIONS
|
# Mock PREFERRED_SESSIONS
|
||||||
with patch('session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]):
|
with patch('src.session_manager.PREFERRED_SESSIONS', [(2, "18:30", "CONDITIONING")]):
|
||||||
result = session_manager.matches_preferred_session(session, current_time)
|
result = session_manager.matches_preferred_session(session, current_time)
|
||||||
assert result is False
|
assert result is False
|
||||||
@@ -12,7 +12,7 @@ from unittest.mock import patch, mock_open
|
|||||||
import sys
|
import sys
|
||||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
|
|
||||||
from session_config import SessionConfig
|
from src.session_config import SessionConfig
|
||||||
|
|
||||||
class TestSessionConfig:
|
class TestSessionConfig:
|
||||||
|
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ from unittest.mock import patch, MagicMock
|
|||||||
import sys
|
import sys
|
||||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
|
|
||||||
from session_notifier import SessionNotifier
|
from src.session_notifier import SessionNotifier
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def email_credentials():
|
def email_credentials():
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ from dotenv import load_dotenv
|
|||||||
import sys
|
import sys
|
||||||
|
|
||||||
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||||
from session_notifier import SessionNotifier
|
from src.session_notifier import SessionNotifier
|
||||||
|
|
||||||
# Load environment variables from .env file
|
# Load environment variables from .env file
|
||||||
load_dotenv()
|
load_dotenv()
|
||||||
|
|||||||
Reference in New Issue
Block a user