diff --git a/booking.md b/booking.md new file mode 100644 index 0000000..7b3050c --- /dev/null +++ b/booking.md @@ -0,0 +1,44 @@ +MAIN: + +- Book court C of sport S at club X at time T for users U1,U2 + +* X.has_ongoing_bookings(U1) +* X.book(C, S, T, U1) + +- Cancel booking B at club X for user U1 + +* X.cancel(B, U1) + +- Get tournaments of sport S at club X + +* X.get_tournaments(S) + +Club: + +- Book court C of sport S at time T for users U1,U2 + +* new ClubUserSession + +- Cancel booking B for user U1 +- Has user U1 ongoing booking + +ClubConnector: + +- land +- login user U +- book court C of sport S at time T +- has ongoing bookings +- cancel booking B +- get tournaments + +UserSession + ++ user ++ session ++ connector + +- book + +* connector.land +* connector.login +* connector.land diff --git a/resa_padel/booking.py b/resa_padel/booking.py index 2a0b7ec..7ac2576 100644 --- a/resa_padel/booking.py +++ b/resa_padel/booking.py @@ -2,62 +2,86 @@ import asyncio import logging import config -from gestion_sports.gestion_sports_platform import GestionSportsPlatform -from models import BookingFilter, Club, User +from connectors import Connector, GestionSportsConnector +from models import Action, BookingFilter, Club, Court, User LOGGER = logging.getLogger(__name__) -async def book(club: Club, user: User, booking_filter: BookingFilter) -> int | None: +def get_connector(club: Club) -> Connector: + if club.booking_platform.id == "gestion-sports": + return GestionSportsConnector(club) + + +async def book_court( + club: Club, users: list[User], booking_filter: BookingFilter +) -> tuple[Court, User]: """ - Book a court for a user to a club following a booking filter + Book any court that meets the condition from the filter. IThe user that will make + the booking is chosen among a list of users and should not have any ongoing bookings - :param club: the club where to book a court - :param user: the user information - :param booking_filter: the information related to the booking - :return: the id of the booked court, or None if no court was booked + :param club: the club in which the booking will be made + :param users: the list of users who have an account in the club + :param booking_filter: the conditions the court to book should meet + :return: a tuple containing the court that was booked and the user who made the + booking """ - async with GestionSportsPlatform(club) as platform: - return await platform.book(user, booking_filter) + connector = get_connector(club) + for user in users: + if not await connector.has_user_ongoing_booking(user): + return await connector.book(user, booking_filter), user -async def get_user_without_booking(club: Club, users: list[User]) -> User | None: +async def cancel_booking(club: Club, user: User, booking_filter: BookingFilter) -> None: """ - Return the first user who has no booking + Cancel the booking that matches the specified filter - :param club: the club where to book - :param users: the list of users - :return: any user who has no booking + :param club: the club in which the booking was made + :param user: the user who made the booking + :param booking_filter: the conditions to meet to cancel the booking """ - async with GestionSportsPlatform(club) as platform: - for user in users: - if await platform.user_has_no_ongoing_booking(user): - return user - return None + connector = get_connector(club) + await connector.cancel_booking(user, booking_filter) -def main() -> int | None: +async def cancel_booking_id(club: Club, user: User, booking_id: int) -> None: + """ + Cancel a booking that matches the booking id + + :param club: the club in which the booking was made + :param user: the user who made the booking + :param booking_id: the id of the booking to cancel + """ + connector = get_connector(club) + await connector.cancel_booking_id(user, booking_id) + + +def main() -> tuple[Court, User] | None: """ Main function used to book a court :return: the id of the booked court, or None if no court was booked """ - booking_filter = config.get_booking_filter() - club = config.get_club() - user = asyncio.run(get_user_without_booking(club, config.get_available_users())) + action = config.get_action() - LOGGER.info( - "Starting booking court of sport %s for user %s at club %s at %s", - booking_filter.sport_id, - user.login, - club.id, - booking_filter.date, - ) - court_booked = asyncio.run(book(club, user, booking_filter)) - if court_booked: - LOGGER.info( - "Court %s booked successfully at %s", court_booked, booking_filter.date - ) - else: - LOGGER.info("Booking did not work") - return court_booked + if action == Action.BOOK: + club = config.get_club() + users = config.get_users(club.id) + booking_filter = config.get_booking_filter() + court_booked, user = asyncio.run(book_court(club, users, booking_filter)) + if court_booked: + LOGGER.info( + "Court %s booked successfully at %s for user %s", + court_booked, + booking_filter.date, + user, + ) + return court_booked, user + else: + LOGGER.info("Booking did not work") + + elif action == Action.CANCEL: + user = config.get_user() + club = config.get_club() + booking_filter = config.get_booking_filter() + asyncio.run(cancel_booking(club, user, booking_filter)) diff --git a/resa_padel/booking_service.py b/resa_padel/booking_service.py new file mode 100644 index 0000000..1c9c437 --- /dev/null +++ b/resa_padel/booking_service.py @@ -0,0 +1,42 @@ +import logging + +from aiohttp import ClientSession +from connectors import Connector +from models import BookingFilter, Club, User + +LOGGER = logging.getLogger(__name__) + + +class BookingService: + def __init__(self, club: Club, connector: Connector): + LOGGER.info("Initializing booking service at for club", club.name) + self.club = club + self.connector = connector + self.session: ClientSession | None = None + + async def __aenter__(self): + self.session = ClientSession() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.session.close() + + async def book(self, user: User, booking_filter: BookingFilter) -> int | None: + """ + Book a court matching the booking filters for a user. + The steps to perform a booking are to go to the landing page, to log in, wait + and for the time when booking is open and then actually book the court + + :param user: the user that wants to book a court + :param booking_filter: the booking criteria + :return: the court number if the booking is successful, None otherwise + """ + if self.connector is None: + LOGGER.error("No connection to Gestion Sports is available") + return None + + if user is None or booking_filter is None: + LOGGER.error("Not enough information available to book a court") + return None + + self.connector.book(user, booking_filter) diff --git a/resa_padel/config.py b/resa_padel/config.py index 749ea5e..a0069ae 100644 --- a/resa_padel/config.py +++ b/resa_padel/config.py @@ -6,36 +6,11 @@ from pathlib import Path import pendulum import yaml from dotenv import load_dotenv -from models import BookingFilter, Club, User +from models import Action, BookingFilter, Club, User load_dotenv() - -def get_club() -> Club: - """ - Read the environment variables related to the current club - and build the Club object - - :return: the club - """ - club_url = os.environ.get("CLUB_URL") - court_ids_tmp = os.environ.get("COURT_IDS") or "" - court_ids = ( - [int(court_id) for court_id in court_ids_tmp.split(",")] - if court_ids_tmp - else [] - ) - club_id = os.environ.get("CLUB_ID") - booking_open_days_before = int(os.environ.get("BOOKING_OPEN_DAYS_BEFORE", "7")) - booking_opening_time_str = os.environ.get("BOOKING_OPENING_TIME", "00:00") - booking_opening_time = pendulum.parse(booking_opening_time_str) - return Club( - id=club_id, - url=club_url, - courts_ids=court_ids, - booking_open_days_before=booking_open_days_before, - booking_opening_time=booking_opening_time.time(), - ) +ROOT_DIR = Path(__file__).parent def get_booking_filter() -> BookingFilter: @@ -45,11 +20,10 @@ def get_booking_filter() -> BookingFilter: :return: the club """ - sport_id_tmp = os.environ.get("SPORT_ID") - sport_id = int(sport_id_tmp) if sport_id_tmp else None + sport_name = os.environ.get("SPORT_NAME") date_time_tmp = os.environ.get("DATE_TIME") date_time = pendulum.parse(date_time_tmp) if date_time_tmp else None - return BookingFilter(sport_id=sport_id, date=date_time) + return BookingFilter(sport_name=sport_name.lower(), date=date_time) def get_user() -> User: @@ -64,25 +38,6 @@ def get_user() -> User: return User(login=login, password=password) -def get_available_users() -> list[User]: - """ - Read the environment variables to get all the available users in order - to increase the chance of having a user with a free slot for a booking - - :return: the list of all users that can book a court - """ - available_users_credentials = os.environ.get("AVAILABLE_USERS_CREDENTIALS") - available_users = [ - credential for credential in available_users_credentials.split(",") - ] - users = [] - for user in available_users: - login, password = user.split(":") - users.append(User(login=login, password=password)) - - return users - - def get_post_headers(platform_id: str) -> dict: """ Get the headers for the POST endpoint related to a specific booking platform @@ -102,13 +57,90 @@ def init_log_config(): """ Read the logging.yaml file to initialize the logging configuration """ - root_dir = os.path.realpath(os.path.dirname(__file__)) - logging_file = root_dir + "/logging.yaml" + logging_file = ROOT_DIR / "logging.yaml" - with open(logging_file, "r") as f: + with logging_file.open(mode="r", encoding="utf-8") as f: logging_config = yaml.safe_load(f.read()) - logging.config.dictConfig(logging_config) + + logging.config.dictConfig(logging_config) -ROOT_PATH = Path(__file__).parent.resolve() -RESOURCES_DIR = Path(ROOT_PATH, "resources") +def _build_urls(platform_urls: dict) -> dict: + return {url["name"]: url for url in platform_urls} + + +def _read_clubs(platforms_data: dict, clubs_data: dict) -> dict[str, Club]: + platforms = {platform["id"]: platform for platform in platforms_data} + + for club in clubs_data["clubs"]: + club_platform = club["bookingPlatform"] + platform_id = club_platform["id"] + club_platform["urls"] = _build_urls(platforms[platform_id]["urls"]) + return {club["id"]: Club(**club) for club in clubs_data["clubs"]} + + +def get_clubs(): + platforms_file = ROOT_DIR / "resources" / "platforms.yaml" + with platforms_file.open(mode="r", encoding="utf-8") as fp: + platforms_data = yaml.safe_load(fp) + + clubs_file = ROOT_DIR / "resources" / "clubs.yaml" + with clubs_file.open(mode="r", encoding="utf-8") as fp: + clubs_data = yaml.safe_load(fp) + + return _read_clubs(platforms_data["platforms"], clubs_data) + + +def get_club() -> Club: + """ + Get the club from an environment variable + + :return: the club + """ + club_id = os.environ.get("CLUB_ID") + clubs = get_clubs() + return clubs[club_id] + + +def read_users(data: dict, club_id: str) -> list[User]: + """ + Deserialize users + + :param data: the dictionnary of users + :param club_id: the club id + :return: a list if users from the club + """ + for club in data.get("clubs"): + if club.get("id") == club_id: + return [User(**user) for user in club.get("users")] + + +def get_users(club_id: str) -> list[User]: + """ + Get a list of users from a club + + :param club_id: the club to which the users should have an account + :return: the list of all users for that club + """ + users_file = ROOT_DIR / "resources" / "users.yaml" + with users_file.open(mode="r", encoding="utf-8") as fp: + data = yaml.safe_load(fp) + + return read_users(data, club_id) + + +def get_resources_folder() -> Path: + """ + Compute the path to the resources used by the program + :return: the path to the resources folder + """ + default_resources_folder = Path(__file__).parent / "resources" + return Path(os.environ.get("RESOURCES_FOLDER", default_resources_folder)) + + +def get_action() -> Action: + """ + Get the action to perform from an environment variable + :return: the action to perform + """ + return Action(os.environ.get("ACTION")) diff --git a/resa_padel/connectors.py b/resa_padel/connectors.py new file mode 100644 index 0000000..c36149e --- /dev/null +++ b/resa_padel/connectors.py @@ -0,0 +1,598 @@ +import asyncio +import json +import logging +import time +from abc import ABC, abstractmethod +from pathlib import Path +from urllib.parse import urljoin + +import config +import pendulum +from aiohttp import ClientResponse, ClientSession +from bs4 import BeautifulSoup +from models import Booking, BookingFilter, Club, Court, Sport, User +from payload_builders import PayloadBuilder +from pendulum import DateTime + +LOGGER = logging.getLogger(__name__) + +POST_HEADERS = config.get_post_headers("gestion-sports") + + +class Connector(ABC): + """ + Abstract class that defines the method a connector + to a website for sport booking should have + """ + + @abstractmethod + async def book(self, user: User, booking_filter: BookingFilter) -> Court | None: + """ + Book a court matching the filter for a user + + :param user: the user who will have the booking + :param booking_filter: the conditions to book (date, time, court) + :return: the court booked + """ + pass + + @abstractmethod + async def has_user_ongoing_booking(self, user: User) -> bool: + """ + Test whether the user has ongoing bookings + + :param user: the user who will have the booking + :return: true if the user has at least one ongoing booking, false otherwise + """ + pass + + @abstractmethod + async def cancel_booking_id(self, user: User, booking_id: int) -> None: + """ + Cancel the booking for a given user + + :param user: the user who has the booking + :param booking_id: the id of the booking + """ + pass + + @abstractmethod + async def cancel_booking(self, user: User, booking_filter: BookingFilter) -> None: + """ + Cancel the booking that meet some conditions for a given user + + :param user: the user who has the booking + :param booking_filter: the booking conditions to meet to cancel the booking + """ + pass + + +class GestionSportsConnector(Connector): + """ + The connector for the Gestion Sports platform. + It handles all the requests to the website. + """ + + def __init__(self, club: Club): + if club is None: + raise ValueError("A connector cannot be instantiated without a club") + if club.booking_platform.id != "gestion-sports": + raise ValueError( + "Gestion Sports connector was instantiated with a club not handled" + " by gestions sports. Club id is {} instead of gestion-sports".format( + club.club_id + ) + ) + + self.club = club + + def _get_url_path(self, name: str) -> str: + """ + Get the URL path for the service with the given name + + :param name: the name of the service + :return: the URL path + """ + self._check_url_path_exists(name) + + return urljoin( + self.club.booking_platform.url, + self.club.booking_platform.urls.get(name).path, + ) + + def _get_payload_template(self, name: str) -> Path: + """ + Get the path to the template file for the service with the given name + + :param name: the name of the service + :return: the path to the template file + """ + self._check_payload_template_exists(name) + + return ( + config.get_resources_folder() + / self.club.booking_platform.urls.get(name).payload_template + ) + + def _check_url_path_exists(self, name: str) -> None: + """ + Check that the URL path for the given service is defined + + :param name: the name of the service + """ + if ( + self.club.booking_platform.urls is None + or self.club.booking_platform.urls.get(name) is None + or self.club.booking_platform.urls.get(name).path is None + ): + raise ValueError( + f"The booking platform internal URL path for page {name} of club " + f"{self.club.name} are not set" + ) + + def _check_payload_template_exists(self, name: str) -> None: + """ + Check that the payload template for the given service is defined + + :param name: the name of the service + """ + if ( + self.club.booking_platform.urls is None + or self.club.booking_platform.urls.get(name) is None + or self.club.booking_platform.urls.get(name).path is None + ): + raise ValueError( + f"The booking platform internal URL path for page {name} of club " + f"{self.club.name} are not set" + ) + + @property + def landing_url(self) -> str: + """ + Get the URL to the landing page of Gestion-Sports + + :return: the URL to the landing page + """ + return self._get_url_path("landing-page") + + @property + def login_url(self) -> str: + """ + Get the URL to the connection login of Gestion-Sports + + :return: the URL to the login page + """ + return self._get_url_path("login") + + @property + def login_template(self) -> Path: + """ + Get the payload template to send to log in the website + + :return: the payload template for logging in + """ + return self._get_payload_template("login") + + @property + def booking_url(self) -> str: + """ + Get the URL to the booking page of Gestion-Sports + + :return: the URL to the booking page + """ + return self._get_url_path("booking") + + @property + def booking_template(self) -> Path: + """ + Get the payload template to send to book a court + + :return: the payload template for booking a court + """ + return self._get_payload_template("booking") + + @property + def user_bookings_url(self) -> str: + """ + Get the URL where all the user's bookings are available + + :return: the URL to the user's bookings + """ + return self._get_url_path("user-bookings") + + @property + def user_bookings_template(self) -> Path: + """ + Get the payload template to send to get all the user's bookings that are + available + + :return: the payload template for the user's bookings + """ + return self._get_payload_template("user-bookings") + + @property + def booking_cancellation_url(self) -> str: + """ + Get the URL where all the user's bookings are available + + :return: the URL to the user's bookings + """ + return self._get_url_path("cancellation") + + @property + def booking_cancellation_template(self) -> Path: + """ + Get the payload template to send to get all the user's bookings that are + available + + :return: the payload template for the user's bookings + """ + return self._get_payload_template("cancellation") + + @property + def available_sports(self) -> dict[str, Sport]: + """ + Get a dictionary of all sports, the key is the sport name lowered case + :return: the dictionary of all sports + """ + return { + sport.name.lower(): sport for sport in self.club.booking_platform.sports + } + + async def land(self, session: ClientSession) -> ClientResponse: + """ + Perform the request to the landing page in order to get the cookie PHPSESSIONID + + :return: the response from the landing page + """ + LOGGER.info("Connecting to GestionSports API at %s", self.login_url) + async with session.get(self.landing_url) as response: + await response.text() + return response + + async def login(self, session: ClientSession, user: User) -> ClientResponse: + """ + Perform the request to the log in the user + + :return: the response from the login + """ + LOGGER.info("Logging in to GestionSports API at %s", self.login_url) + payload = PayloadBuilder.build(self.login_template, user=user, club=self.club) + + async with session.post( + self.login_url, data=payload, headers=POST_HEADERS + ) as response: + resp_text = await response.text() + LOGGER.debug("Connexion request response:\n%s", resp_text) + return response + + async def book(self, user: User, booking_filter: BookingFilter) -> Court | None: + """ + Perform a request for each court at the same time to increase the chances to get + a booking. + The gestion-sports backend does not allow several bookings at the same time + so there is no need to make each request one after the other + + :param user: the user that wants to book the court + :param booking_filter: the booking conditions to meet + :return: the booked court, or None if no court was booked + """ + LOGGER.info( + "Booking any available court from GestionSports API at %s", self.booking_url + ) + + sport = self.available_sports.get(booking_filter.sport_name) + + async with ClientSession() as session: + # use asyncio to request a booking on every court + # the gestion-sports backend is able to book only one court for a user + await self.land(session) + await self.login(session, user) + self.wait_until_booking_time(booking_filter) + bookings = await asyncio.gather( + *[ + self.send_booking_request( + session, booking_filter.date, court.id, sport.id + ) + for court in sport.courts + ], + return_exceptions=True, + ) + + LOGGER.debug("Booking results:\n'%s'", bookings) + return self.get_booked_court(bookings, sport.name) + + async def send_booking_request( + self, + session: ClientSession, + date: DateTime, + court_id: int, + sport_id: int, + ) -> tuple[ClientResponse, int, bool]: + """ + Book a single court that meets the conditions from the booking filter + + :param session: the HTTP session that contains the user information and cookies + :param date: the booking date + :param court_id: the id of the court to book + :param sport_id: the id of the sport + :return: a tuple containing the court id and the booking status + """ + LOGGER.debug("Booking court %s at %s", court_id, date.to_w3c_string()) + payload = PayloadBuilder.build( + self.booking_template, + date=date, + court_id=court_id, + sport_id=sport_id, + ) + + LOGGER.debug("Booking court %s request:\n'%s'", court_id, payload) + + async with session.post( + self.booking_url, data=payload, headers=POST_HEADERS + ) as response: + assert response.status == 200 + resp_json = await response.text() + LOGGER.debug("Response from booking request:\n'%s'", resp_json) + return response, court_id, self.is_booking_response_status_ok(resp_json) + + def get_booked_court( + self, bookings: list[tuple[ClientSession, int, bool]], sport_name: str + ) -> Court | None: + """ + Parse the booking list and return the court that was booked + + :param bookings: a list of bookings + :param sport_name: the sport name + :return: the id of the booked court if any, None otherwise + """ + for _, court_id, is_booked in bookings: + if is_booked: + LOGGER.debug("Court %d is booked", court_id) + court_booked = self.find_court(court_id, sport_name) + LOGGER.info("Court '%s' is booked", court_booked.name) + return court_booked + LOGGER.debug("No booked court found") + return None + + def find_court(self, court_id: int, sport_name: str) -> Court: + """ + Get all the court information based on the court id and the sport name + + :param court_id: the court id + :param sport_name: the sport name + :return: the court that has the given id and sport name + """ + sport = self.available_sports.get(sport_name.lower()) + for court in sport.courts: + if court.id == court_id: + return court + + @staticmethod + def is_booking_response_status_ok(response: str) -> bool: + """ + Check if the booking response is OK + + :param response: the response as a string + :return: true if the status is ok, false otherwise + """ + formatted_result = response.removeprefix('"').removesuffix('"') + result_json = json.loads(formatted_result) + return result_json["status"] == "ok" + + def build_booking_datetime(self, booking_filter: BookingFilter) -> DateTime: + """ + Build the date and time when the booking is open for a given match date. + The booking filter contains the date and time of the booking. + The club has the information about when the booking is open for that date. + + :param booking_filter: the booking information + :return: the date and time when the booking is open + """ + date_to_book = booking_filter.date + booking_opening = self.club.booking_platform.booking_opening + booking_date = date_to_book.subtract(days=booking_opening.days_before) + + opening_time = pendulum.parse(booking_opening.opening_time) + booking_hour = opening_time.hour + booking_minute = opening_time.minute + + return booking_date.at(booking_hour, booking_minute) + + def wait_until_booking_time(self, booking_filter: BookingFilter) -> None: + """ + Wait until the booking is open. + The booking filter contains the date and time of the booking. + The club has the information about when the booking is open for that date. + + :param booking_filter: the booking information + """ + LOGGER.info("Waiting for booking time") + booking_datetime = self.build_booking_datetime(booking_filter) + now = pendulum.now() + duration_until_booking = booking_datetime - now + LOGGER.debug(f"Current time: {now}, Datetime to book: {booking_datetime}") + LOGGER.debug( + f"Time to wait before booking: {duration_until_booking.hours:0>2}" + f":{duration_until_booking.minutes:0>2}" + f":{duration_until_booking.seconds:0>2}" + ) + + while now < booking_datetime: + time.sleep(1) + now = pendulum.now() + LOGGER.info("It's booking time!") + + async def has_user_ongoing_booking(self, user: User) -> bool: + """ + Check if the user currently has bookings in the future + :param user: the user to check the bookings + :return: true if the user has some bookings, false otherwise + """ + async with ClientSession() as session: + await self.land(session) + await self.login(session, user) + return bool(await self.get_ongoing_bookings(session)) + + async def get_ongoing_bookings(self, session: ClientSession) -> list[Booking]: + """ + Get the list of all ongoing bookings of a user. + The steps to perform this are to get the user's bookings page and get a hidden + property in the HTML to get a hash that will be used in the payload of the + POST request (sic) to get the user's bookings. + Gestion sports is really a mess!! + + :return: the list of all ongoing bookings of a user + """ + hash_value = await self.send_hash_request(session) + LOGGER.debug(f"Hash value: {hash_value}") + payload = PayloadBuilder.build(self.user_bookings_template, hash=hash_value) + LOGGER.debug(f"Payload to get ongoing bookings: {payload}") + return await self.send_user_bookings_request(session, payload) + + async def send_hash_request(self, session: ClientSession) -> str: + """ + Get the hash value used in the request to get the user's bookings + + :param session: the session in which the user logged in + :return: the value of the hash + """ + async with session.get(self.user_bookings_url) as response: + html = await response.text() + LOGGER.debug("Get bookings response: %s\n", html) + return self.get_hash_input(html) + + @staticmethod + def get_hash_input(html_doc: str) -> str: + """ + There is a secret hash generated by Gestion sports that is reused when trying to get + users bookings. This hash is stored in a hidden input with name "mesresas-hash" + + :param html_doc: the html document when getting the page mes-resas.html + :return: the value of the hash in the page + """ + soup = BeautifulSoup(html_doc, "html.parser") + inputs = soup.find_all("input") + for input_tag in inputs: + if input_tag.get("name") == "mesresas-hash": + return input_tag.get("value").strip() + + async def send_user_bookings_request( + self, session: ClientSession, payload: str + ) -> list[Booking]: + """ + Perform the HTTP request to get all bookings + + :param session: the session in which the user logged in + :param payload: the HTTP payload for the request + :return: a dictionary containing all the bookings + """ + async with session.post( + self.user_bookings_url, data=payload, headers=POST_HEADERS + ) as response: + resp = await response.text() + LOGGER.debug("ongoing bookings response: %s\n", resp) + return [Booking(**booking) for booking in json.loads(resp)] + + async def cancel_booking_id(self, user: User, booking_id: int) -> ClientResponse: + """ + Cancel a booking based on its id for a given user + + :param user: the user that has the booking + :param booking_id: the id of the booking to cancel + :return: the response from the client + """ + async with ClientSession() as session: + await self.land(session) + await self.login(session, user) + + return await self.send_cancellation_request(session, booking_id) + + async def send_cancellation_request( + self, session: ClientSession, booking_id: int + ) -> ClientResponse: + """ + Send the HTTP request to cancel the booking + + :param session: the HTTP session that contains the user information and cookies + :param booking_id: the id of the booking to cancel + :return: the response from the client + """ + hash_value = await self.send_hash_request(session) + payload = PayloadBuilder.build( + self.booking_cancellation_template, + booking_id=booking_id, + hash=hash_value, + ) + + async with session.post( + self.booking_cancellation_url, data=payload, headers=POST_HEADERS + ) as response: + await response.text() + return response + + async def cancel_booking(self, user: User, booking_filter: BookingFilter) -> None: + """ + Cancel the booking that meets some conditions + + :param user: the user who owns the booking + :param booking_filter: the conditions the booking to cancel should meet + """ + async with ClientSession() as session: + await self.land(session) + await self.login(session, user) + + bookings = await self.get_ongoing_bookings(session) + + for booking in bookings: + if self.is_booking_matching_filter(booking, booking_filter): + await self.send_cancellation_request(session, booking.id) + + def is_booking_matching_filter( + self, booking: Booking, booking_filter: BookingFilter + ) -> bool: + """ + Check if the booking matches the booking filter + + :param booking: the booking to be checked + :param booking_filter: the conditions the booking should meet + :return: true if the booking matches the conditions, false otherwise + """ + return ( + self._is_same_sport(booking, booking_filter) + and self._is_date_matching(booking, booking_filter) + and self._is_time_matching(booking, booking_filter) + ) + + @staticmethod + def _is_same_sport(booking: Booking, booking_filter: BookingFilter) -> bool: + """ + Check if the booking and the booking filter are about the same sport + + :param booking: the booking to be checked + :param booking_filter: the conditions the booking should meet + :return: true if the booking sport matches the filter sport, false otherwise + """ + return booking.sport == booking_filter.sport_name + + @staticmethod + def _is_date_matching(booking: Booking, booking_filter: BookingFilter) -> bool: + """ + Check if the booking and the booking filter are at the same date + + :param booking: the booking to be checked + :param booking_filter: the conditions the booking should meet + :return: true if the booking date matches the filter date, false otherwise + """ + return booking.booking_date.date() == booking_filter.date.date() + + @staticmethod + def _is_time_matching(booking: Booking, booking_filter: BookingFilter) -> bool: + """ + Check if the booking and the booking filter are at the same time + + :param booking: the booking to be checked + :param booking_filter: the conditions the booking should meet + :return: true if the booking time matches the filter time, false otherwise + """ + return booking.start_time.time() == booking_filter.date.time() diff --git a/resa_padel/gestion_sports/__init__.py b/resa_padel/gestion_sports/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/resa_padel/gestion_sports/gestion_sports_config.py b/resa_padel/gestion_sports/gestion_sports_config.py deleted file mode 100644 index ea5b852..0000000 --- a/resa_padel/gestion_sports/gestion_sports_config.py +++ /dev/null @@ -1,9 +0,0 @@ -from pathlib import Path - -import config - -RESOURCES_DIR = Path(config.RESOURCES_DIR, "gestion-sports") - -BOOKING_TEMPLATE = Path(RESOURCES_DIR, "booking-payload.txt") -LOGIN_TEMPLATE = Path(RESOURCES_DIR, "login-payload.txt") -USERS_BOOKINGS_TEMPLATE = Path(RESOURCES_DIR, "users_bookings.txt") diff --git a/resa_padel/gestion_sports/gestion_sports_connector.py b/resa_padel/gestion_sports/gestion_sports_connector.py deleted file mode 100644 index de5dca3..0000000 --- a/resa_padel/gestion_sports/gestion_sports_connector.py +++ /dev/null @@ -1,199 +0,0 @@ -import asyncio -import json -import logging -from urllib.parse import urljoin - -import config -from aiohttp import ClientResponse, ClientSession -from gestion_sports import gestion_sports_html_parser as html_parser -from gestion_sports.payload_builders import ( - GestionSportsBookingPayloadBuilder, - GestionSportsLoginPayloadBuilder, - GestionSportsUsersBookingsPayloadBuilder, -) -from models import BookingFilter, Club, User - -LOGGER = logging.getLogger(__name__) -POST_HEADERS = config.get_post_headers("gestion-sports") - - -class GestionSportsConnector: - """ - Handle the specific booking requests to Gestion-Sports - """ - - def __init__(self, session: ClientSession, url: str): - LOGGER.info("Initializing connection to GestionSports API") - self.url = url - self.session = session - - @property - def landing_url(self) -> str: - """ - Get the URL to the landing page of Gestion-Sports - - :return: the URL to the landing page - """ - return urljoin(self.url, "/connexion.php") - - @property - def login_url(self) -> str: - """ - Get the URL to the connection login of Gestion-Sports - - :return: the URL to the login page - """ - return urljoin(self.url, "/connexion.php") - - @property - def booking_url(self) -> str: - """ - Get the URL to the booking page of Gestion-Sports - - :return: the URL to the booking page - """ - return urljoin(self.url, "/membre/reservation.html") - - @property - def user_bookings_url(self) -> str: - """ - Get the URL where all the user's bookings are available - - :return: the URL to the user's bookings - """ - return urljoin(self.url, "/membre/mesresas.html") - - async def land(self) -> ClientResponse: - """ - Perform the request to the landing page in order to get the cookie PHPSESSIONID - - :return: the response from the landing page - """ - LOGGER.info("Connecting to GestionSports API at %s", self.login_url) - async with self.session.get(self.landing_url) as response: - await response.text() - return response - - async def login(self, user: User, club: Club) -> ClientResponse: - """ - Perform the request to the log in the user - - :return: the response from the login - """ - LOGGER.info("Logging in to GestionSports API at %s", self.login_url) - payload_builder = GestionSportsLoginPayloadBuilder() - payload = payload_builder.user(user).club(club).build() - - async with self.session.post( - self.login_url, data=payload, headers=POST_HEADERS - ) as response: - resp_text = await response.text() - LOGGER.debug("Connexion request response:\n%s", resp_text) - return response - - async def book(self, booking_filter: BookingFilter, club: Club) -> int | None: - """ - Perform a request for each court at the same time to increase the chances to get - a booking. - The gestion-sports backend does not allow several bookings at the same time - so there is no need to make each request one after the other - - :param booking_filter: the booking information - :param club: the club where to book the court - :return: the booked court, or None if no court was booked - """ - LOGGER.info( - "Booking any available court from GestionSports API at %s", self.booking_url - ) - # use asyncio to request a booking on every court - # the gestion-sports backend is able to book only one court for a user - bookings = await asyncio.gather( - *[ - self.book_one_court(booking_filter, court_id) - for court_id in club.courts_ids - ], - return_exceptions=True, - ) - - LOGGER.debug("Booking results:\n'%s'", bookings) - return self.get_booked_court(bookings) - - async def book_one_court( - self, booking_filter: BookingFilter, court_id: int - ) -> tuple[int, bool]: - """ - Book a single court according to the information provided in the booking filter - - :param booking_filter: the booking information - :param court_id: the id of the court to book - :return: a tuple containing the court id and the booking status - """ - LOGGER.debug( - "Booking court %s at %s", - court_id, - booking_filter.date.to_w3c_string(), - ) - payload_builder = GestionSportsBookingPayloadBuilder() - payload = ( - payload_builder.booking_filter(booking_filter).court_id(court_id).build() - ) - LOGGER.debug("Booking court %s request:\n'%s'", court_id, payload) - - async with self.session.post( - self.booking_url, data=payload, headers=POST_HEADERS - ) as response: - resp_json = await response.text() - LOGGER.debug("Response from booking request:\n'%s'", resp_json) - return court_id, self.is_booking_response_status_ok(resp_json) - - @staticmethod - def get_booked_court(bookings: list[tuple[int, bool]]) -> int | None: - """ - Parse the booking list and return the court that was booked - - :param bookings: a list of bookings - :return: the id of the booked court if any, None otherwise - """ - for court, is_booked in bookings: - if is_booked: - LOGGER.debug("Court %s is booked", court) - return court - LOGGER.debug("No booked court found") - return None - - @staticmethod - def is_booking_response_status_ok(response: str) -> bool: - """ - Check if the booking response is OK - - :param response: the response as a string - :return: true if the status is ok, false otherwise - """ - formatted_result = response.removeprefix('"').removesuffix('"') - result_json = json.loads(formatted_result) - return result_json["status"] == "ok" - - async def get_ongoing_bookings(self) -> dict: - """ - Get the list of all ongoing bookings of a user. - The steps to perform this are to get the user's bookings page and get a hidden - property in the HTML to get a hash that will be used in the payload of the - POST request (sic) to get the user's bookings. - Gestion sports is really a mess!! - - :return: the list of all ongoing bookings of a user - """ - async with self.session.get(self.user_bookings_url) as get_resp: - html = await get_resp.text() - hash_value = html_parser.get_hash_input(html) - - payload_builder = GestionSportsUsersBookingsPayloadBuilder() - payload_builder.hash(hash_value) - payload = payload_builder.build() - - async with self.session.post( - self.user_bookings_url, data=payload, headers=POST_HEADERS - ) as response: - resp = await response.text() - LOGGER.debug("ongoing bookings response: %s\n", resp) - return json.loads(resp) diff --git a/resa_padel/gestion_sports/gestion_sports_html_parser.py b/resa_padel/gestion_sports/gestion_sports_html_parser.py deleted file mode 100644 index ed7f12a..0000000 --- a/resa_padel/gestion_sports/gestion_sports_html_parser.py +++ /dev/null @@ -1,16 +0,0 @@ -from bs4 import BeautifulSoup - - -def get_hash_input(html_doc: str) -> str: - """ - There is a secret hash generated by Gestion sports that is reused when trying to get - users bookings. This hash is stored in a hidden input with name "mesresas-hash" - - :param html_doc: the html document when getting the page mes-resas.html - :return: the value of the hash in the page - """ - soup = BeautifulSoup(html_doc, "html.parser") - inputs = soup.find_all("input") - for input_tag in inputs: - if input_tag.get("name") == "mesresas-hash": - return input_tag.get("value").strip() diff --git a/resa_padel/gestion_sports/gestion_sports_platform.py b/resa_padel/gestion_sports/gestion_sports_platform.py deleted file mode 100644 index 35679ce..0000000 --- a/resa_padel/gestion_sports/gestion_sports_platform.py +++ /dev/null @@ -1,116 +0,0 @@ -import logging -import time - -import pendulum -from aiohttp import ClientSession -from gestion_sports.gestion_sports_connector import GestionSportsConnector -from models import BookingFilter, Club, User -from pendulum import DateTime - -LOGGER = logging.getLogger(__name__) - - -class GestionSportsPlatform: - def __init__(self, club: Club): - LOGGER.info("Initializing Gestion Sports platform at url %s", club.url) - self.connector: GestionSportsConnector | None = None - self.club: Club = club - self.session: ClientSession | None = None - - async def __aenter__(self): - self.session = ClientSession() - self.connector = GestionSportsConnector(self.session, self.club.url) - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb): - await self.session.close() - - async def book(self, user: User, booking_filter: BookingFilter) -> int | None: - """ - Book a court matching the booking filters for a user. - The steps to perform a booking are to go to the landing page, to log in, wait - and for the time when booking is open and then actually book the court - - :param user: the user that wants to book a court - :param booking_filter: the booking criteria - :return: the court number if the booking is successful, None otherwise - """ - if self.connector is None: - LOGGER.error("No connection to Gestion Sports is available") - return None - - if user is None or booking_filter is None: - LOGGER.error("Not enough information available to book a court") - return None - - await self.connector.land() - await self.connector.login(user, self.club) - wait_until_booking_time(self.club, booking_filter) - return await self.connector.book(booking_filter, self.club) - - async def user_has_no_ongoing_booking(self, user: User) -> bool: - """ - Check if the user has any ongoing booking. - The steps to perform this task are to go to the landing page, to log in and - then retrieve user information and extract the ongoing bookings - - :param user: the user to check the bookings - :return: True if the user has ongoing bookings, false otherwise - """ - if self.connector is None: - LOGGER.error("No connection to Gestion Sports is available") - return False - - if user is None: - LOGGER.error("Not enough information available to book a court") - return False - - await self.connector.land() - await self.connector.login(user, self.club) - bookings = await self.connector.get_ongoing_bookings() - return bookings == [] - - -def wait_until_booking_time(club: Club, booking_filter: BookingFilter): - """ - Wait until the booking is open. - The booking filter contains the date and time of the booking. - The club has the information about when the booking is open for that date. - - :param club: the club where to book a court - :param booking_filter: the booking information - """ - LOGGER.info("Waiting for booking time") - booking_datetime = build_booking_datetime(booking_filter, club) - now = pendulum.now() - duration_until_booking = booking_datetime - now - LOGGER.debug( - "Time to wait before booking: %s:%s:%s", - "{:0>2}".format(duration_until_booking.hours), - "{:0>2}".format(duration_until_booking.minutes), - "{:0>2}".format(duration_until_booking.seconds), - ) - - while now < booking_datetime: - time.sleep(1) - now = pendulum.now() - LOGGER.info("It's booking time!") - - -def build_booking_datetime(booking_filter: BookingFilter, club: Club) -> DateTime: - """ - Build the date and time when the booking is open for a given match date. - The booking filter contains the date and time of the booking. - The club has the information about when the booking is open for that date. - - :param booking_filter: the booking information - :param club: the club where to book a court - :return: the date and time when the booking is open - """ - date_to_book = booking_filter.date - booking_date = date_to_book.subtract(days=club.booking_open_days_before) - - booking_hour = club.booking_opening_time.hour - booking_minute = club.booking_opening_time.minute - - return booking_date.at(booking_hour, booking_minute) diff --git a/resa_padel/gestion_sports/payload_builders.py b/resa_padel/gestion_sports/payload_builders.py deleted file mode 100644 index 1509a52..0000000 --- a/resa_padel/gestion_sports/payload_builders.py +++ /dev/null @@ -1,128 +0,0 @@ -from exceptions import ArgumentMissing -from gestion_sports.gestion_sports_config import ( - BOOKING_TEMPLATE, - LOGIN_TEMPLATE, - USERS_BOOKINGS_TEMPLATE, -) -from jinja2 import Environment, FileSystemLoader -from models import BookingFilter, Club, User - - -class GestionSportsLoginPayloadBuilder: - """ - Build the payload for the login page - """ - - def __init__(self): - self._user: User | None = None - self._club: Club | None = None - - def user(self, user: User): - """ - Set the user - - :param user: the user - :return: the class itself - """ - self._user = user - return self - - def club(self, club: Club): - """ - Set the club - - :param club: the club - :return: the class itself - """ - self._club = club - return self - - def build(self) -> str: - """ - Build the payload - - :return: the string representation of the payload - """ - if self._user is None: - raise ArgumentMissing("No user was provided") - if self._club is None: - raise ArgumentMissing("No club was provided") - - environment = Environment(loader=FileSystemLoader(LOGIN_TEMPLATE.parent)) - template = environment.get_template(LOGIN_TEMPLATE.name) - - return template.render(club=self._club, user=self._user) - - -class GestionSportsBookingPayloadBuilder: - def __init__(self): - self._booking_filter: BookingFilter | None = None - self._court_id: int | None = None - - def booking_filter(self, booking_filter: BookingFilter): - """ - Set the booking filter - - :param booking_filter: the booking filter - :return: the class itself - """ - self._booking_filter = booking_filter - return self - - def court_id(self, court_id: int): - """ - Set the court id - - :param court_id: the court id - :return: the class itself - """ - self._court_id = court_id - return self - - def build(self) -> str: - """ - Build the payload - - :return: the string representation of the payload - """ - if self._booking_filter is None: - raise ArgumentMissing("No booking filter was provided") - if self.court_id is None: - raise ArgumentMissing("No court id was provided") - - environment = Environment(loader=FileSystemLoader(BOOKING_TEMPLATE.parent)) - template = environment.get_template(BOOKING_TEMPLATE.name) - - return template.render( - court_id=self._court_id, booking_filter=self._booking_filter - ) - - -class GestionSportsUsersBookingsPayloadBuilder: - def __init__(self): - self._hash: str | None = None - - def hash(self, hash_value: str): - """ - Set the hash - - :param hash_value: the hash - :return: the class itself - """ - self._hash = hash_value - - def build(self) -> str: - """ - Build the payload - - :return: the string representation of the payload - """ - if self._hash is None: - raise ArgumentMissing("No hash was provided") - - environment = Environment( - loader=FileSystemLoader(USERS_BOOKINGS_TEMPLATE.parent) - ) - template = environment.get_template(USERS_BOOKINGS_TEMPLATE.name) - - return template.render(hash=self._hash) diff --git a/resa_padel/models.py b/resa_padel/models.py index 1857750..6045abd 100644 --- a/resa_padel/models.py +++ b/resa_padel/models.py @@ -1,22 +1,171 @@ -from pendulum import Time -from pydantic import BaseModel, ConfigDict, Field +from enum import Enum +from typing import Optional + +import pendulum +from pydantic import BaseModel, ConfigDict, Field, field_validator from pydantic_extra_types.pendulum_dt import DateTime -class Club(BaseModel): +class User(BaseModel): + login: str + password: str = Field(repr=False) + club_id: Optional[str] = Field(default=None) + + +class BookingOpening(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) - id: str = Field() - url: str = Field() - courts_ids: list[int] = Field(default_factory=list) - booking_open_days_before: int = Field(default=7) - booking_opening_time: Time = Field(default=Time(hour=0, minute=0)) + + days_before: Optional[int] = Field(default=7, alias="daysBefore") + opening_time: Optional[str] = Field(alias="time", default=None, repr=False) + time_after_booking: Optional[str] = Field( + alias="timeAfterBookingTime", default=None, repr=False + ) + + def __repr__(self): + base = super().__repr__() + time = f", time: {self.time})" if self.time else "" + time_after_booking = ( + f", time_after_booking_time: {self.time_after_booking_time})" + if self.time_after_booking_time + else "" + ) + + return base.removesuffix(")") + time + time_after_booking + + @property + def time(self): + return pendulum.parse(self.opening_time).time() + + @property + def time_after_booking_time(self): + return ( + pendulum.parse(self.time_after_booking).time() + if self.time_after_booking + else None + ) + + +class TotalBookings(BaseModel): + peak_hours: int | str = Field(alias="peakHours") + off_peak_hours: int | str = Field(alias="offPeakHours") + + +class Court(BaseModel): + id: int + name: str + number: int + is_indoor: Optional[bool] = Field(alias="isIndoor") + + +class Sport(BaseModel): + name: str + id: int + duration: int + price: int + players: int + courts: list[Court] + + +class Url(BaseModel): + name: str + path: str + payload_template: Optional[str] = Field(default=None, alias="payloadTemplate") + + +class BookingPlatform(BaseModel): + id: str + club_id: int = Field(alias="clubId") + url: str + hours_before_cancellation: int = Field(alias="hoursBeforeCancellation") + booking_opening: BookingOpening = Field(alias="bookingOpening") + total_bookings: TotalBookings = Field(alias="totalBookings") + sports: list[Sport] + urls: dict[str, Url] + + +class Club(BaseModel): + id: str + name: str + url: str + booking_platform: BookingPlatform = Field(alias="bookingPlatform") + + +class PlatformDefinition(BaseModel): + id: str + name: str + url: str + urls: list[Url] class BookingFilter(BaseModel): - sport_id: int = Field() - date: DateTime = Field() + date: DateTime + sport_name: str + + @field_validator("sport_name", mode="before") + @classmethod + def to_lower_case(cls, d: str) -> str: + return d.lower() -class User(BaseModel): - login: str = Field() - password: str = Field(repr=False) +class Booking(BaseModel): + id: int + booking_date: DateTime = Field(alias="dateResa") + start_time: DateTime = Field(alias="startTime") + sport: str + court: str + game_creation: Optional[int] = Field(default=None, alias="creaPartie") + game_creation_limit: Optional[DateTime] = Field( + default=None, alias="limitCreaPartie" + ) + cancel: Optional[bool] = Field(default=True) + block_player_replacement: Optional[int] = Field( + default=None, alias="bloquerRemplacementJoueur" + ) + can_remove_parteners: bool = Field(default=True, alias="canRemovePartners") + end_time: Optional[DateTime] = Field(default=None, alias="endTime") + day_fr: Optional[str] = Field(default=None, alias="dayFr") + live_xperience_code: Optional[str] = Field(default=None, alias="codeLiveXperience") + spartime_qr_code: Optional[str] = Field(default=None, alias="qrCodeSpartime") + remaining_places: int = Field(default=3, alias="remainingPlaces") + is_captain: bool = Field(default=True, alias="isCaptain") + dt_start: Optional[DateTime] = Field(default=None, alias="dtStart") + credit_card_guaranty: Optional[str] = Field(default=None, alias="garantieCb") + certificate_validity_duration: Optional[int] = Field( + alias="dureeValidCertif", default=None + ) + charge_id: Optional[str] = Field(default=None, alias="chargeId") + partners: Optional[list] = Field(default=[]) + player_status: Optional[int] = Field(default=None, alias="playerStatus") + products: Optional[list] = Field(default=[]) + + @field_validator("booking_date", mode="before") + @classmethod + def validate_date(cls, d: str) -> DateTime: + return pendulum.from_format( + d, "DD/MM/YYYY", tz=pendulum.timezone("Europe/Paris") + ) + + @field_validator("start_time", "end_time", mode="before") + @classmethod + def validate_time(cls, t: str) -> DateTime: + return pendulum.from_format(t, "HH:mm", tz=pendulum.timezone("Europe/Paris")) + + @field_validator("game_creation_limit", mode="before") + @classmethod + def validate_datetime_add_tz(cls, dt: str) -> DateTime: + return pendulum.parse(dt, tz=pendulum.timezone("Europe/Paris")) + + @field_validator("dt_start", mode="before") + @classmethod + def validate_datetime(cls, dt: str) -> DateTime: + return pendulum.parse(dt) + + @field_validator("sport", mode="before") + @classmethod + def to_lower_case(cls, d: str) -> str: + return d.lower() + + +class Action(Enum): + BOOK = "book" + CANCEL = "cancel" diff --git a/resa_padel/payload_builders.py b/resa_padel/payload_builders.py new file mode 100644 index 0000000..5c5c338 --- /dev/null +++ b/resa_padel/payload_builders.py @@ -0,0 +1,12 @@ +from pathlib import Path + +from jinja2 import Environment, FileSystemLoader + + +class PayloadBuilder: + @staticmethod + def build(template_path: Path, **kwargs) -> str: + environment = Environment(loader=FileSystemLoader(template_path.parent)) + template = environment.get_template(template_path.name) + + return template.render(**kwargs) diff --git a/resa_padel/resources/clubs.yaml b/resa_padel/resources/clubs.yaml new file mode 100644 index 0000000..0689fda --- /dev/null +++ b/resa_padel/resources/clubs.yaml @@ -0,0 +1,160 @@ +clubs: + - name: Toulouse Padel Club + url: https://www.toulousepadelclub.com + id: tpc + bookingPlatform: + id: gestion-sports + clubId: 88 + url: https://toulousepadelclub.gestion-sports.com + hoursBeforeCancellation: 4 + bookingOpening: + daysBefore: 7 + time: 00:00 + totalBookings: + peakHours: 1 + offPeakHours: unlimited + sports: + - name: Padel + id: 217 + duration: 90 + price: 48 + players: 4 + courts: + - name: Court 1 + number: 1 + id: 598 + isIndoor: True + - name: Court 2 + number: 2 + id: 599 + isIndoor: True + - name: Court 3 + number: 3 + id: 600 + isIndoor: True + - name: Court 4 + number: 4 + id: 601 + isIndoor: True + - name: Court 5 + number: 5 + id: 602 + isIndoor: True + - name: Court 6 + number: 6 + id: 603 + isIndoor: True + - name: Court 7 + number: 7 + id: 604 + isIndoor: True + - name: Court 8 + number: 8 + id: 605 + isIndoor: True + - name: Court 9 + number: 9 + id: 606 + isIndoor: True + - name: Court 10 + number: 10 + id: 607 + isIndoor: True + - name: Court 11 + number: 11 + id: 608 + isIndoor: True + - name: Court 12 + number: 12 + id: 609 + isIndoor: True + - name: Court 13 + number: 13 + id: 610 + isIndoor: True + - name: Court 14 + number: 14 + id: 611 + isIndoor: True + - name: Squash + id: 218 + duration: 45 + price: 18 + players: 2 + courts: + - name: Court 1 + id: 613 + number: 1 + isIndoor: True + - name: Court 2 + number: 2 + id: 614 + isIndoor: True + - name: Court 3 + number: 3 + id: 615 + isIndoor: True + - name: Court 4 + number: 4 + id: 616 + isIndoor: True + + - name: Padel Tolosa + url: https://www.padeltolosa.fr/ + id: padeltolosa + bookingPlatform: + id: gestion-sports + clubId: 89 + url: https://padeltolosa.gestion-sports.com/ + hoursBeforeCancellation: 24 + bookingOpening: + daysBefore: 7 + time: 00:00 + totalBookings: + peakHours: 4 + offPeakHours: unlimited + sports: + - name: Padel + id: 262 + duration: 90 + price: 48 + players: 4 + courts: + - name: Court 1 M.F IMMOBILLIER + number: 1 + id: 746 + isIndoor: True + - name: Court 2 PAQUITO + number: 2 + id: 747 + isIndoor: True + - name: Court 3 Seven Sisters PUB TOULOUSE + number: 3 + id: 748 + isIndoor: True + - name: Court MADRID + number: 4 + id: 749 + isIndoor: True + - name: Squash + id: 218 + duration: 45 + price: 18 + players: 2 + courts: + - name: Court 1 + id: 613 + isIndoor: True + number: 1 + - name: Court 2 + id: 614 + isIndoor: True + number: 2 + - name: Court 3 + id: 615 + isIndoor: True + number: 3 + - name: Court 4 + id: 616 + isIndoor: True + number: 4 diff --git a/resa_padel/resources/gestion-sports/booking-cancellation-payload.txt b/resa_padel/resources/gestion-sports/booking-cancellation-payload.txt new file mode 100644 index 0000000..2083999 --- /dev/null +++ b/resa_padel/resources/gestion-sports/booking-cancellation-payload.txt @@ -0,0 +1 @@ +ajax=removeResa&hash={{ hash }}&id={{ booking_id }} diff --git a/resa_padel/resources/gestion-sports/booking-payload.txt b/resa_padel/resources/gestion-sports/booking-payload.txt index b5a7cc5..c416a11 100644 --- a/resa_padel/resources/gestion-sports/booking-payload.txt +++ b/resa_padel/resources/gestion-sports/booking-payload.txt @@ -1 +1 @@ -ajax=addResa&date={{ booking_filter.date.date().strftime("%d/%m/%Y") }}&hour={{ booking_filter.date.time().strftime("%H:%M") }}&duration=90&partners=null|null|null&paiement=facultatif&idSport={{ booking_filter.sport_id }}&creaPartie=false&idCourt={{ court_id }}&pay=false&token=undefined&totalPrice=44&saveCard=0&foodNumber=0 +ajax=addResa&date={{ date.date().strftime("%d/%m/%Y") }}&hour={{ date.time().strftime("%H:%M") }}&duration=90&partners=null|null|null&paiement=facultatif&idSport={{ sport_id }}&creaPartie=false&idCourt={{ court_id }}&pay=false&token=undefined&totalPrice=48&saveCard=0&foodNumber=0 diff --git a/resa_padel/resources/gestion-sports/login-payload.txt b/resa_padel/resources/gestion-sports/login-payload.txt index b0da065..7692aed 100644 --- a/resa_padel/resources/gestion-sports/login-payload.txt +++ b/resa_padel/resources/gestion-sports/login-payload.txt @@ -1 +1 @@ -ajax=connexionUser&id_club={{ club.id }}&email={{ user.login }}&form_ajax=1&pass={{ user.password }}&compte=user&playeridonesignal=0&identifiant=identifiant&externCo=true +ajax=connexionUser&id_club={{ club.booking_platform.club_id }}&email={{ user.login }}&form_ajax=1&pass={{ user.password }}&compte=user&playeridonesignal=0&identifiant=identifiant&externCo=true diff --git a/resa_padel/resources/gestion-sports/users_bookings.txt b/resa_padel/resources/gestion-sports/user-bookings-payload.txt similarity index 100% rename from resa_padel/resources/gestion-sports/users_bookings.txt rename to resa_padel/resources/gestion-sports/user-bookings-payload.txt diff --git a/resa_padel/resources/platforms.yaml b/resa_padel/resources/platforms.yaml new file mode 100644 index 0000000..7f1525f --- /dev/null +++ b/resa_padel/resources/platforms.yaml @@ -0,0 +1,19 @@ +platforms: + - name: Gestion sports + url: https://gestion-sports.fr/ + id: gestion-sports + urls: + - name: landing-page + path: /connexion.php + - name: login + path: /connexion.php + payloadTemplate: gestion-sports/login-payload.txt + - name: booking + path: /membre/reservation.html + payloadTemplate: gestion-sports/booking-payload.txt + - name: user-bookings + path: /membre/mesresas.html + payloadTemplate: gestion-sports/user-bookings-payload.txt + - name: cancellation + path: /membre/mesresas.html + payloadTemplate: gestion-sports/booking-cancellation-payload.txt diff --git a/resa_padel/resources/users.yaml b/resa_padel/resources/users.yaml new file mode 100644 index 0000000..1d53d16 --- /dev/null +++ b/resa_padel/resources/users.yaml @@ -0,0 +1,13 @@ +clubs: + - id: tpc + users: + - login: padel.testing@jouf.fr + password: ridicule + - login: mateochal31@gmail.com + password: pleanyakis + - id: padeltolosa + users: + - login: padel.testing@jouf.fr + password: ridicule + - login: mateochal31@gmail.com + password: pleanyakis diff --git a/tests/data/user_bookings_html_response.html b/tests/data/user_bookings_html_response.html new file mode 100644 index 0000000..3af1e3b --- /dev/null +++ b/tests/data/user_bookings_html_response.html @@ -0,0 +1,1363 @@ + + + + + + Mes réservations + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+ +
+
+ +
+
+ + + Retour + + + + + +
+ +
+ Mes réservations
+
+ + + 6 + +
+
+ + + +
+
+
+ + +
+
+
Mes événements
+
+
+
+
+
+ +
+
Mes réservations
+
+
+
+ + + Réserver un court + +
+
+
+
+
+
Suivi de vos réservations par sport
+
+ + Heure pleines + Heure creuses +
+
+ Padel + +
+
0 /1
+
0 /1
+
+ + +
+
illimitées
+
illimitées
+
+ +
+
+ Squash + +
+
0 /2
+
0 /2
+
+ + +
+
illimitées
+
illimitées
+
+ +
+
+ Electrostimulation + +
+
illimitées
+
illimitées
+
+ + +
+
illimitées
+
illimitées
+
+ +
+
+
+ + +
+ + + + + + + +
+ +
+ Copyright © Gestion-Sports 2024. Tous droits réservés. +
+
+ + + + + + + +
+ + + +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+ +
+ + Accueil +
+ + + +
+ + Actualités +
+ + + +
+
+ + Réserver +
+
+ + + + +
+ + Compte +
+ + + +
+ + Menu +
+ + +
+ + + + + + + + + + + + + + + + + + +
+
+
+
+ image + Le Mas + à l'instant +
+ + + +
+
+
+

Titre de la notif

+
+ Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod + tempor incididunt ut labore et dolore magna aliqua. +
+
+
+
+
+ + + + + + + + + + + +
+
+ +
+ Votre inscription à la partie de Tennis a bien été enregistrée !
+ Rendez-vous le 14 Janvier de 16h30 à 17h30. +
+
+ +
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/tests/fixtures.py b/tests/fixtures.py index bef8442..c7d514f 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -3,19 +3,16 @@ from pathlib import Path import pendulum import pytest -from gestion_sports.payload_builders import GestionSportsBookingPayloadBuilder -from resa_padel.models import BookingFilter, Club, User +from resa_padel.models import BookingFilter, User user = User(login="padel.testing@jouf.fr", password="ridicule", club_id="123") -url = "https://tpc.padel.com" -club = Club(id="123", url=url, courts_ids=[606, 607, 608]) courts = [606, 607, 608] -sport_id = 217 +sport_name = "padel" tz_info = "Europe/Paris" booking_date = pendulum.now().add(days=6).set(hour=18, minute=0, second=0, tz=tz_info) -booking_filter = BookingFilter(sport_id=sport_id, date=booking_date) +booking_filter = BookingFilter(sport_name=sport_name, date=booking_date) booking_failure_response = json.dumps( { @@ -36,12 +33,6 @@ booking_success_response = json.dumps( date_format = "%d/%m/%Y" time_format = "%H:%M" -booking_payload = ( - GestionSportsBookingPayloadBuilder() - .booking_filter(booking_filter) - .court_id(courts[0]) - .build() -) html_file = Path(__file__).parent / "data" / "mes_resas.html" _mes_resas_html = html_file.read_text(encoding="utf-8") @@ -57,11 +48,6 @@ def a_booking_filter() -> BookingFilter: return booking_filter -@pytest.fixture -def a_club() -> Club: - return club - - @pytest.fixture def a_booking_success_response() -> str: return booking_success_response @@ -72,11 +58,6 @@ def a_booking_failure_response() -> str: return booking_failure_response -@pytest.fixture -def a_booking_payload() -> str: - return booking_payload - - @pytest.fixture def mes_resas_html() -> str: return _mes_resas_html diff --git a/tests/gestion_sports/test_gestion_sports_payload_builder.py b/tests/gestion_sports/test_gestion_sports_payload_builder.py index 4b7d23d..b494c55 100644 --- a/tests/gestion_sports/test_gestion_sports_payload_builder.py +++ b/tests/gestion_sports/test_gestion_sports_payload_builder.py @@ -1,8 +1,9 @@ -from resa_padel.gestion_sports.payload_builders import ( +from payload_builders import ( GestionSportsBookingPayloadBuilder, GestionSportsLoginPayloadBuilder, GestionSportsUsersBookingsPayloadBuilder, ) + from tests.fixtures import a_booking_filter, a_club, a_user diff --git a/tests/test_booking.py b/tests/test_booking.py index b60ae4d..4de9767 100644 --- a/tests/test_booking.py +++ b/tests/test_booking.py @@ -1,77 +1,70 @@ +import asyncio import os from unittest.mock import patch +import config import pendulum -from aioresponses import aioresponses -from models import BookingFilter, Club -from pendulum import Time +from models import BookingFilter, User from resa_padel import booking -from tests import fixtures, utils -from tests.fixtures import ( - a_booking_failure_response, - a_booking_success_response, - mes_resas_html, + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, ) +def test_real_booking(): + club = config.get_club() + user = User(login="padel.testing@jouf.fr", password="ridicule") + booking_filter = BookingFilter( + sport_name="Padel", date=pendulum.parse("2024-03-21T13:30:00+01:00") + ) + booked_court, user_that_booked = asyncio.run( + booking.book_court(club, [user], booking_filter) + ) + assert booked_court is not None + assert user_that_booked == user -login = "user" -password = "password" -available_credentials = login + ":" + password + ",some_user:some_password" -club_id = "88" -court_id = "11" -paris_tz = "Europe/Paris" -datetime_to_book = ( - pendulum.now().add(days=6).set(hour=18, minute=0, second=0, tz=paris_tz) + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, ) +def test_real_cancellation(): + club = config.get_club() + user = User(login="padel.testing@jouf.fr", password="ridicule") + asyncio.run(booking.cancel_booking_id(club, user, 3605033)) -@patch("pendulum.now") @patch.dict( os.environ, { - "LOGIN": login, - "PASSWORD": password, - "CLUB_ID": club_id, - "CLUB_URL": fixtures.url, - "COURT_IDS": "7,8,10", - "SPORT_ID": "217", - "DATE_TIME": datetime_to_book.isoformat(), - "AVAILABLE_USERS_CREDENTIALS": available_credentials, + "CLUB_ID": "tpc", + "ACTION": "book", + "SPORT_NAME": "Padel", + "DATE_TIME": "2024-03-21T13:30:00+01:00", }, clear=True, ) -def test_main( - mock_now, - a_booking_success_response: str, - a_booking_failure_response: str, - mes_resas_html: str, -): - """ - Test the main function to book a court +def test_main_booking(): + court, user = booking.main() + assert court is not None + assert user.username == "padel.testing@jouf" - :param mock_now: the pendulum.now() mock - :param a_booking_success_response: the success json response - :param a_booking_failure_response: the failure json response - """ - booking_filter = BookingFilter(sport_id=666, date=datetime_to_book) - club = Club( - id="club", - url="some.url", - courts_ids=[7, 8, 10], - booking_open_days_before=7, - booking_opening_time=Time(hour=0, minute=0), - ) - booking_datetime = utils.retrieve_booking_datetime(booking_filter, club) - mock_now.side_effect = [booking_datetime] - with aioresponses() as aio_mock: - utils.mock_rest_api_from_connection_to_booking( - aio_mock, - fixtures.url, - a_booking_failure_response, - a_booking_success_response, - mes_resas_html, - ) - - court_booked = booking.main() - assert court_booked == 8 +@patch.dict( + os.environ, + { + "CLUB_ID": "tpc", + "ACTION": "cancel", + "SPORT_NAME": "Padel", + "DATE_TIME": "2024-03-21T13:30:00+01:00", + "LOGIN": "padel.testing@jouf.fr", + "PASSWORD": "ridicule", + }, + clear=True, +) +def test_main_cancellation(): + booking.main() diff --git a/tests/test_config.py b/tests/test_config.py index 0c1e25b..c9c76ef 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -2,40 +2,20 @@ import os from unittest.mock import patch import config -from pendulum import DateTime, Time, Timezone +from pendulum import DateTime, Timezone @patch.dict( os.environ, { - "CLUB_URL": "club.url", - "COURT_IDS": "7,8,10", - "CLUB_ID": "666", - "BOOKING_OPEN_DAYS_BEFORE": "5", - "BOOKING_OPENING_TIME": "18:37", - }, - clear=True, -) -def test_get_club(): - club = config.get_club() - assert club.url == "club.url" - assert club.courts_ids == [7, 8, 10] - assert club.id == "666" - assert club.booking_open_days_before == 5 - assert club.booking_opening_time == Time(hour=18, minute=37) - - -@patch.dict( - os.environ, - { - "SPORT_ID": "666", + "SPORT_NAME": "Padel", "DATE_TIME": "2024-02-03T22:38:45Z", }, clear=True, ) def test_get_booking_filter(): booking_filter = config.get_booking_filter() - assert booking_filter.sport_id == 666 + assert booking_filter.sport_id == "padel" assert booking_filter.date == DateTime( year=2024, month=2, @@ -61,14 +41,11 @@ def test_get_available_user(): assert user.password == "gloups" -@patch.dict( - os.environ, - {"AVAILABLE_USERS_CREDENTIALS": "login@user.tld:gloups,other@user.tld:patatras"}, - clear=True, -) -def test_user(): - users = config.get_available_users() - assert users[0].login == "login@user.tld" - assert users[0].password == "gloups" - assert users[1].login == "other@user.tld" - assert users[1].password == "patatras" +def test_read_clubs(): + clubs = config.get_clubs() + assert len(clubs) == 2 + + +def test_get_users(): + users = config.get_users("tpc") + assert len(users) == 2 diff --git a/tests/test_connectors.py b/tests/test_connectors.py new file mode 100644 index 0000000..9f503ed --- /dev/null +++ b/tests/test_connectors.py @@ -0,0 +1,411 @@ +import os +from pathlib import Path +from unittest.mock import patch + +import aiohttp +import config +import pendulum +import pytest +from aiohttp import ClientSession +from connectors import GestionSportsConnector +from models import Booking, BookingFilter, User +from yarl import URL + +from tests import utils + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +def test_urls(): + club = config.get_club() + connector = GestionSportsConnector(club) + assert ( + connector.landing_url + == "https://toulousepadelclub.gestion-sports.com/connexion.php" + ) + assert ( + connector.login_url + == "https://toulousepadelclub.gestion-sports.com/connexion.php" + ) + assert ( + connector.booking_url + == "https://toulousepadelclub.gestion-sports.com/membre/reservation.html" + ) + assert ( + connector.user_bookings_url + == "https://toulousepadelclub.gestion-sports.com/membre/mesresas.html" + ) + assert ( + connector.booking_cancellation_url + == "https://toulousepadelclub.gestion-sports.com/membre/mesresas.html" + ) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc", "RESOURCES_FOLDER": "/some/path"}, + clear=True, +) +def test_urls_payload_templates(): + club = config.get_club() + connector = GestionSportsConnector(club) + resources_folder = Path("/some", "path", "gestion-sports") + assert connector.login_template == resources_folder / "login-payload.txt" + assert connector.booking_template == resources_folder / "booking-payload.txt" + assert ( + connector.user_bookings_template + == resources_folder / "user-bookings-payload.txt" + ) + assert ( + connector.booking_cancellation_template + == resources_folder / "booking-cancellation-payload.txt" + ) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_landing_page(): + club = config.get_club() + connector = GestionSportsConnector(club) + async with aiohttp.ClientSession() as session: + response = await connector.land(session) + + assert response.status == 200 + assert response.request_info.method == "GET" + assert response.content_type == "text/html" + assert response.request_info.url == URL(connector.landing_url) + assert response.charset == "UTF-8" + assert response.cookies.get("PHPSESSID") is not None + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_login(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + + async with aiohttp.ClientSession() as session: + await connector.land(session) + + response = await connector.login(session, user) + + assert response.status == 200 + assert response.request_info.method == "POST" + assert response.content_type == "text/html" + assert response.request_info.url == URL(connector.landing_url) + assert response.charset == "UTF-8" + assert response.cookies.get("COOK_COMPTE") is not None + assert response.cookies.get("COOK_ID_CLUB").value == "88" + assert response.cookies.get("COOK_ID_USER").value == "232382" + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +def test_get_booked_court(): + club = config.get_club() + connector = GestionSportsConnector(club) + + session = ClientSession() + bookings = [ + (session, 601, False), + (session, 602, False), + (session, 603, False), + (session, 614, False), + (session, 605, False), + (session, 606, True), + (session, 607, False), + (session, 608, False), + ] + + court = connector.get_booked_court(bookings, "padel") + assert court.number == 9 + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_book_one_court(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + + async with aiohttp.ClientSession() as session: + await connector.land(session) + await connector.login(session, user) + response, court_id, ok = await connector.send_booking_request( + session, pendulum.parse("2024-03-21T13:30:00Z"), 610, 217 + ) + + assert response.status == 200 + assert response.request_info.method == "POST" + assert response.content_type == "text/html" + assert response.request_info.url == URL(connector.booking_url) + assert response.charset == "UTF-8" + assert response.text is not None + assert court_id == 610 + assert ok is True + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_book(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + booking_filter = BookingFilter( + sport_name="Padel", date=pendulum.parse("2024-03-21T13:30:00Z") + ) + + booked_court = await connector.book(user, booking_filter) + + assert booked_court is not None + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +def test_build_booking_datetime(): + club = config.get_club() + connector = GestionSportsConnector(club) + booking_filter = BookingFilter( + sport_name="Padel", date=pendulum.parse("2024-03-21T13:30:00Z") + ) + + opening_datetime = connector.build_booking_datetime(booking_filter) + assert opening_datetime.year == 2024 + assert opening_datetime.month == 3 + assert opening_datetime.day == 14 + assert opening_datetime.hour == 0 + assert opening_datetime.minute == 0 + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@patch("pendulum.now") +def test_wait_until_booking_time(mock_now): + club = config.get_club() + connector = GestionSportsConnector(club) + booking_filter = BookingFilter( + sport_name="Padel", date=pendulum.parse("2024-03-21T13:30:00Z") + ) + + booking_datetime = utils.retrieve_booking_datetime(booking_filter, club) + + seconds = [ + booking_datetime.subtract(seconds=3), + booking_datetime.subtract(seconds=2), + booking_datetime.subtract(seconds=1), + booking_datetime, + booking_datetime.add(microseconds=1), + booking_datetime.add(microseconds=2), + ] + mock_now.side_effect = seconds + + connector.wait_until_booking_time(booking_filter) + + assert pendulum.now() == booking_datetime.add(microseconds=1) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_get_hash(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + + async with aiohttp.ClientSession() as session: + await connector.land(session) + await connector.login(session, user) + + hash_value = await connector.send_hash_request(session) + assert hash_value is not None + + +def test_get_hash_input(): + resources_folder = Path(__file__).parent / "data" + html_file = resources_folder / "user_bookings_html_response.html" + html = html_file.read_text(encoding="utf-8") + + hash_value = GestionSportsConnector.get_hash_input(html) + + assert hash_value == "63470fa38e300fd503de1ee21a71b3bdb6fb206b" + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_get_bookings(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + + async with aiohttp.ClientSession() as session: + await connector.land(session) + await connector.login(session, user) + + hash_value = await connector.send_hash_request(session) + payload = f"ajax=loadResa&hash={hash_value}" + + bookings = await connector.send_user_bookings_request(session, payload) + print(bookings) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_get_ongoing_bookings(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + + async with aiohttp.ClientSession() as session: + await connector.land(session) + await connector.login(session, user) + + bookings = await connector.get_ongoing_bookings(session) + print(bookings) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_has_user_ongoing_bookings(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + + assert await connector.has_user_ongoing_booking(user) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_cancel_booking_id(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + + async with aiohttp.ClientSession() as session: + await connector.land(session) + await connector.login(session, user) + ongoing_bookings = await connector.get_ongoing_bookings(session) + booking_id = ongoing_bookings[0].id + + response = await connector.cancel_booking_id(user, booking_id) + + assert len(await connector.get_ongoing_bookings(session)) == 0 + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +def test_is_booking_matching_filter(): + club = config.get_club() + connector = GestionSportsConnector(club) + filter_date = pendulum.parse("2024-03-02T15:00:00+01:00") + booking = Booking( + id=1, + dateResa="02/03/2024", + startTime="15:00", + sport="Padel", + court="10", + ) + booking_filter = BookingFilter(date=filter_date, sport_name="Padel") + + assert connector.is_booking_matching_filter(booking, booking_filter) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +def test_is_booking_not_matching_filter(): + club = config.get_club() + connector = GestionSportsConnector(club) + filter_date = pendulum.parse("2024-03-02T15:00:00+01:00") + booking = Booking( + id=1, + dateResa="02/03/2024", + startTime="16:00", + sport="Padel", + court="10", + ) + booking_filter = BookingFilter(date=filter_date, sport_name="Padel") + + assert not connector.is_booking_matching_filter(booking, booking_filter) + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +def test_find_court(): + club = config.get_club() + connector = GestionSportsConnector(club) + + court = connector.find_court(603, "Padel") + + assert court.number == 6 + + +@patch.dict( + os.environ, + {"CLUB_ID": "tpc"}, + clear=True, +) +@pytest.mark.asyncio +async def test_cancel_booking(): + club = config.get_club() + connector = GestionSportsConnector(club) + user = User(login="padel.testing@jouf.fr", password="ridicule") + filter_date = pendulum.parse("2024-03-21T13:30:00+01:00") + booking_filter = BookingFilter(date=filter_date, sport_name="Padel") + await connector.cancel_booking(user, booking_filter) diff --git a/tests/utils.py b/tests/utils.py index c5383bb..c05ff6b 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -1,5 +1,6 @@ from urllib.parse import urljoin +import pendulum from models import BookingFilter, Club from pendulum import DateTime @@ -7,7 +8,6 @@ from tests.fixtures import ( a_booking_failure_response, a_booking_filter, a_booking_success_response, - a_club, mes_resas_html, ) @@ -65,11 +65,13 @@ def retrieve_booking_datetime( :param a_booking_filter: the booking filter that contains the date to book :param a_club: the club which has the number of days before the date and the booking time """ - booking_hour = a_club.booking_opening_time.hour - booking_minute = a_club.booking_opening_time.minute + booking_opening = a_club.booking_platform.booking_opening + opening_time = pendulum.parse(booking_opening.opening_time) + booking_hour = opening_time.hour + booking_minute = opening_time.minute date_to_book = a_booking_filter.date - return date_to_book.subtract(days=a_club.booking_open_days_before).at( + return date_to_book.subtract(days=booking_opening.days_before).at( booking_hour, booking_minute )