stanislas/resa-padel
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Added docstrings

Browse source
This commit is contained in:
Stanislas Jouffroy 2024-02-18 09:16:11 +01:00
parent 5434a74d0f
commit ccd019eb4c
4 changed files with 174 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

32
resa_padel/booking.py
View file

@ -14,6 +14,14 @@ LOGGER = logging.getLogger(__name__)
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 booking time")
booking_datetime = build_booking_datetime(booking_filter, club)
now = pendulum.now()
@ -23,6 +31,15 @@ def wait_until_booking_time(club: Club, booking_filter: BookingFilter):
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)
@ -33,15 +50,28 @@ def build_booking_datetime(booking_filter: BookingFilter, club: Club) -> DateTim
async def book(club: Club, user: User, booking_filter: BookingFilter) -> int | None:
"""
Book a court for a user to a club following a booking filter
: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
"""
async with ClientSession() as session:
platform = GestionSportsConnector(session, club.url)
await platform.connect()
await platform.land()
await platform.login(user, club)
wait_until_booking_time(club, booking_filter)
return await platform.book(booking_filter, club)
def main() -> int | None:
"""
Main function used to book a court
:return: the id of the booked court, or None if no court was booked
"""
user = config.get_user()
booking_filter = config.get_booking_filter()
club = config.get_club()

27
resa_padel/config.py
View file

@ -13,6 +13,12 @@ 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 = (
@ -34,6 +40,12 @@ def get_club() -> Club:
def get_booking_filter() -> BookingFilter:
"""
Read the environment variables related to the current booking filter
and build the BookingFilter object
:return: the club
"""
sport_id_tmp = os.environ.get("SPORT_ID")
sport_id = int(sport_id_tmp) if sport_id_tmp else None
date_time_tmp = os.environ.get("DATE_TIME")
@ -42,12 +54,24 @@ def get_booking_filter() -> BookingFilter:
def get_user() -> User:
"""
Read the environment variables related to the current user
and build the User object
:return: the club
"""
login = os.environ.get("LOGIN")
password = os.environ.get("PASSWORD")
return User(login=login, password=password)
def get_post_headers(platform_id: str) -> dict:
"""
Get the headers for the POST endpoint related to a specific booking platform
:param platform_id: the platform to which the headers apply
:return: the headers as a dictionary
"""
root_path = Path(__file__).parent
headers_file = Path(root_path, "resources", platform_id, "post-headers.json")
with headers_file.open(mode="r", encoding="utf-8") as f:
@ -57,6 +81,9 @@ def get_post_headers(platform_id: str) -> dict:
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"

43
resa_padel/gestion_sports/gestion_sports_connector.py
View file

@ -18,6 +18,9 @@ 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")
@ -26,24 +29,49 @@ class GestionSportsConnector:
self.payload_builder = GestionSportsPayloadBuilder()
@property
def connection_url(self) -> str:
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?")
async def connect(self) -> ClientResponse:
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")
async with self.session.get(self.connection_url) as response:
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
"""
payload = (
self.payload_builder.login(user.login)
.password(user.password)
@ -58,6 +86,15 @@ class GestionSportsConnector:
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
"""
# 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(

93
tests/gestion_sports/test_gestion_sports_connector.py
View file

@ -2,6 +2,7 @@ import pytest
from aiohttp import ClientSession
from yarl import URL
from models import BookingFilter, Club, User
from resa_padel.gestion_sports.gestion_sports_connector import GestionSportsConnector
from tests.fixtures import (
a_booking_failure_response,
@ -16,13 +17,16 @@ tpc_url = "https://toulousepadelclub.gestion-sports.com"
@pytest.mark.asyncio
async def test_should_connect_to_gestion_sports_website():
async def test_should_reach_landing_page_to_gestion_sports_website() -> None:
"""
Test that landing page is reached
"""
async with ClientSession() as session:
cookies = session.cookie_jar.filter_cookies(URL(tpc_url))
assert cookies.get("PHPSESSID") is None
gs_connector = GestionSportsConnector(session, tpc_url)
response = await gs_connector.connect()
response = await gs_connector.land()
assert response.status == 200
assert response.request_info.method == "GET"
@ -35,10 +39,18 @@ async def test_should_connect_to_gestion_sports_website():
@pytest.mark.asyncio
async def test_should_login_to_gestion_sports_website(a_user, a_club):
async def test_should_login_to_gestion_sports_website(
a_user: User, a_club: Club
) -> None:
"""
Test that a user can log in after reaching the landing page
:param a_user: the user that wants to book a court
:param a_club: the club information
"""
async with ClientSession() as session:
gs_connector = GestionSportsConnector(session, tpc_url)
await gs_connector.connect()
await gs_connector.land()
response = await gs_connector.login(a_user, a_club)
@ -53,10 +65,19 @@ async def test_should_login_to_gestion_sports_website(a_user, a_club):
@pytest.mark.asyncio
async def test_booking_url_should_be_reachable(a_user, a_booking_filter, a_club):
async def test_booking_url_should_be_reachable(
a_user: User, a_booking_filter: BookingFilter, a_club: Club
) -> None:
"""
Test that a user can log in the booking platform and book a court
:param a_user: the user that wants to book a court
:param a_booking_filter: the booking information
:param a_club: the club information
"""
async with ClientSession() as session:
gs_connector = GestionSportsConnector(session, tpc_url)
await gs_connector.connect()
await gs_connector.land()
await gs_connector.login(a_user, a_club)
court_booked = await gs_connector.book(a_booking_filter, a_club)
@ -67,11 +88,21 @@ async def test_booking_url_should_be_reachable(a_user, a_booking_filter, a_club)
@pytest.mark.asyncio
async def test_should_book_a_court_from_gestion_sports(
aioresponses,
a_booking_filter,
a_club,
a_booking_success_response,
a_booking_failure_response,
):
a_booking_filter: BookingFilter,
a_club: Club,
a_booking_success_response: str,
a_booking_failure_response: str,
) -> None:
"""
Test that user can reach the landing page, then log in to the platform
and eventually book a court
:param aioresponses: the http response mock
:param a_booking_filter: the booking information
:param a_club: the club information
:param a_booking_success_response: the success response mock
:param a_booking_failure_response: the failure response mock
"""
booking_url = URL(tpc_url + "/membre/reservation.html?")
# first booking request will fail
@ -89,20 +120,40 @@ async def test_should_book_a_court_from_gestion_sports(
assert court_booked == a_club.courts_ids[1]
def test_response_status_should_be_ok(a_booking_success_response):
def test_response_status_should_be_ok(a_booking_success_response: str) -> None:
"""
Test internal method to verify that the success response received by booking
a gestion-sports court is still a JSON with a field 'status' set to 'ok'
:param a_booking_success_response: the success response mock
"""
is_booked = GestionSportsConnector.is_response_status_ok(a_booking_success_response)
assert is_booked
def test_response_status_should_be_not_ok(a_booking_failure_response):
def test_response_status_should_be_not_ok(a_booking_failure_response: str) -> None:
"""
Test internal method to verify that the failure response received by booking
a gestion-sports court is still a JSON with a field 'status' set to 'error'
:param a_booking_failure_response: the failure response mock
"""
is_booked = GestionSportsConnector.is_response_status_ok(a_booking_failure_response)
assert not is_booked
@pytest.mark.asyncio
async def test_court_should_not_be_booked(
aioresponses, a_booking_payload, a_booking_failure_response
):
aioresponses, a_booking_payload: str, a_booking_failure_response: str
) -> None:
"""
Test that no court is booked when there is a failure response
from the booking request
:param aioresponses: the http requests mock
:param a_booking_payload: the payload that is sent for booking
:param a_booking_failure_response: the failure response mock
"""
async with ClientSession() as session:
tpc_connector = GestionSportsConnector(session, tpc_url)
aioresponses.post(
@ -114,8 +165,16 @@ async def test_court_should_not_be_booked(
@pytest.mark.asyncio
async def test_court_should_be_booked(
aioresponses, a_booking_payload, a_booking_success_response
):
aioresponses, a_booking_payload: str, a_booking_success_response: str
) -> None:
"""
Test that a court is booked when there is a success response
from the booking request
:param aioresponses: the http requests mock
:param a_booking_payload: the payload that is sent for booking
:param a_booking_success_response: the success response mock
"""
async with ClientSession() as session:
tpc_connector = GestionSportsConnector(session, tpc_url)