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

created a gestion sports services class that handles the connection while the connector is dedicated to the requests

Browse source
This commit is contained in:
Stanislas Jouffroy 2024-03-20 23:11:43 +01:00
parent bcd8dc0733
commit e6023e0687
12 changed files with 513 additions and 593 deletions
Download patch file Download diff file Expand all files Collapse all files

21
resa_padel/booking.py
View file

@ -2,17 +2,12 @@ import asyncio
import logging
import config
from connectors import Connector, GestionSportsConnector
from gestion_sports_services import GestionSportsServices
from models import Action, BookingFilter, Club, Court, User
LOGGER = logging.getLogger(__name__)
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]:
@ -26,10 +21,10 @@ async def book_court(
:return: a tuple containing the court that was booked and the user who made the
booking
"""
connector = get_connector(club)
service = GestionSportsServices()
for user in users:
if not await connector.has_user_ongoing_booking(user):
return await connector.book(user, booking_filter), user
if not await service.has_user_available_slots(user, club):
return await service.book(club, user, booking_filter), user
async def cancel_booking(club: Club, user: User, booking_filter: BookingFilter) -> None:
@ -40,8 +35,8 @@ async def cancel_booking(club: Club, user: User, booking_filter: BookingFilter)
:param user: the user who made the booking
:param booking_filter: the conditions to meet to cancel the booking
"""
connector = get_connector(club)
await connector.cancel_booking(user, booking_filter)
service = GestionSportsServices()
await service.cancel_booking(user, club, booking_filter)
async def cancel_booking_id(club: Club, user: User, booking_id: int) -> None:
@ -52,8 +47,8 @@ async def cancel_booking_id(club: Club, user: User, booking_id: int) -> None:
: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)
service = GestionSportsServices()
await service.cancel_booking_id(user, club, booking_id)
def main() -> tuple[Court, User] | None:

167
resa_padel/connectors.py
View file

@ -1,13 +1,10 @@
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
@ -19,55 +16,7 @@ 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):
class GestionSportsConnector:
"""
The connector for the Gestion Sports platform.
It handles all the requests to the website.
@ -266,14 +215,16 @@ class GestionSportsConnector(Connector):
LOGGER.debug("Connexion request response:\n%s", resp_text)
return response
async def book(self, user: User, booking_filter: BookingFilter) -> Court | None:
async def book_any_court(
self, session: ClientSession, booking_filter: BookingFilter
) -> list[tuple[int, dict]]:
"""
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 session: the session to use
:param booking_filter: the booking conditions to meet
:return: the booked court, or None if no court was booked
"""
@ -283,12 +234,6 @@ class GestionSportsConnector(Connector):
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(
@ -300,7 +245,7 @@ class GestionSportsConnector(Connector):
)
LOGGER.debug("Booking results:\n'%s'", bookings)
return self.get_booked_court(bookings, sport.name)
return bookings
async def send_booking_request(
self,
@ -308,7 +253,7 @@ class GestionSportsConnector(Connector):
date: DateTime,
court_id: int,
sport_id: int,
) -> tuple[ClientResponse, int, bool]:
) -> tuple[int, dict]:
"""
Book a single court that meets the conditions from the booking filter
@ -316,7 +261,7 @@ class GestionSportsConnector(Connector):
: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
:return: a tuple containing the court id and the response
"""
LOGGER.debug("Booking court %s at %s", court_id, date.to_w3c_string())
payload = PayloadBuilder.build(
@ -332,12 +277,12 @@ class GestionSportsConnector(Connector):
self.booking_url, data=payload, headers=POST_HEADERS
) as response:
assert response.status == 200
resp_json = await response.text()
resp_json = json.loads(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)
return court_id, resp_json
def get_booked_court(
self, bookings: list[tuple[ClientSession, int, bool]], sport_name: str
self, bookings: list[tuple[int, dict]], sport_name: str
) -> Court | None:
"""
Parse the booking list and return the court that was booked
@ -346,8 +291,8 @@ class GestionSportsConnector(Connector):
: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:
for court_id, response in bookings:
if self.is_booking_response_status_ok(response):
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)
@ -369,70 +314,14 @@ class GestionSportsConnector(Connector):
return court
@staticmethod
def is_booking_response_status_ok(response: str) -> bool:
def is_booking_response_status_ok(response: dict) -> 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))
return response["status"] == "ok"
async def get_ongoing_bookings(self, session: ClientSession) -> list[Booking]:
"""
@ -494,21 +383,7 @@ class GestionSportsConnector(Connector):
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(
async def cancel_booking_id(
self, session: ClientSession, booking_id: int
) -> ClientResponse:
"""
@ -533,20 +408,16 @@ class GestionSportsConnector(Connector):
return response
async def cancel_booking(
self, user: User, booking_filter: BookingFilter
self, session: ClientSession, booking_filter: BookingFilter
) -> ClientResponse | None:
"""
Cancel the booking that meets some conditions
:param user: the user who owns the booking
:param session: the session
: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 booking.matches(booking_filter):
return await self.send_cancellation_request(session, booking.id)
return await self.cancel_booking_id(session, booking.id)

127
resa_padel/gestion_sports_services.py Normal file
View file

@ -0,0 +1,127 @@
import logging
import time
import pendulum
from aiohttp import ClientSession
from connectors import GestionSportsConnector
from models import BookingFilter, BookingOpening, Club, Court, User
from pendulum import DateTime
LOGGER = logging.getLogger(__name__)
class GestionSportsServices:
@staticmethod
async def book(
club: Club, 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 club: the club in which the booking will be made
: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
"""
connector = GestionSportsConnector(club)
LOGGER.info(
"Booking any available court from GestionSports API at %s",
connector.booking_url,
)
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 connector.land(session)
await connector.login(session, user)
booking_opening = club.booking_platform.booking_opening
GestionSportsServices.wait_until_booking_time(
booking_filter, booking_opening
)
bookings = await connector.book_any_court(session, booking_filter)
LOGGER.debug("Booking results:\n'%s'", bookings)
return connector.get_booked_court(bookings, booking_filter.sport_name)
@staticmethod
async def has_user_available_slots(user: User, club: Club) -> bool:
connector = GestionSportsConnector(club)
async with ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
bookings = await connector.get_ongoing_bookings(session)
return bool(bookings)
@staticmethod
async def cancel_booking(user: User, club: Club, booking_filter: BookingFilter):
connector = GestionSportsConnector(club)
async with ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
await connector.cancel_booking(session, booking_filter)
@staticmethod
async def cancel_booking_id(user: User, club: Club, booking_id: int):
connector = GestionSportsConnector(club)
async with ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
await connector.cancel_booking_id(session, booking_id)
@staticmethod
def wait_until_booking_time(
booking_filter: BookingFilter, booking_opening: BookingOpening
) -> 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_opening:
:param booking_filter: the booking information
"""
LOGGER.info("Waiting for booking time")
booking_datetime = GestionSportsServices.build_booking_datetime(
booking_filter, booking_opening
)
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!")
@staticmethod
def build_booking_datetime(
booking_filter: BookingFilter, booking_opening: BookingOpening
) -> 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_opening:the booking opening conditions
:param booking_filter: the booking information
:return: the date and time when the booking is open
"""
date_to_book = booking_filter.date
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)

46
tests/integration_tests/conftest.py Normal file
View file

@ -0,0 +1,46 @@
import json
from pathlib import Path
import config
import pendulum
import pytest
from connectors import GestionSportsConnector
from models import BookingFilter, Club, User
TEST_FOLDER = Path(__file__).parent.parent
DATA_FOLDER = TEST_FOLDER / "data"
RESPONSES_FOLDER = DATA_FOLDER / "responses"
@pytest.fixture
def club() -> Club:
return config.get_clubs()["tpc"]
@pytest.fixture
def connector(club) -> GestionSportsConnector:
return GestionSportsConnector(club)
@pytest.fixture
def user() -> User:
return User(login="padel.testing@jouf.fr", password="ridicule")
@pytest.fixture
def booking_filter() -> BookingFilter:
return BookingFilter(
sport_name="Padel", date=pendulum.parse("2024-03-21T13:30:00+01:00")
)
@pytest.fixture
def booking_success_response() -> dict:
booking_success_file = RESPONSES_FOLDER / "booking_success.json"
return json.loads(booking_success_file.read_text(encoding="utf-8"))
@pytest.fixture
def booking_failure_response() -> dict:
booking_failure_file = RESPONSES_FOLDER / "booking_failure.json"
return json.loads(booking_failure_file.read_text(encoding="utf-8"))

21
tests/integration_tests/test_booking.py
View file

@ -2,10 +2,6 @@ import asyncio
import os
from unittest.mock import patch
import config
import pendulum
from models import BookingFilter, User
from resa_padel import booking
@ -14,12 +10,7 @@ from resa_padel import booking
{"CLUB_ID": "tpc"},
clear=True,
)
def test_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")
)
def test_booking(club, user, booking_filter):
booked_court, user_that_booked = asyncio.run(
booking.book_court(club, [user], booking_filter)
)
@ -32,10 +23,8 @@ def test_booking():
{"CLUB_ID": "tpc"},
clear=True,
)
def test_cancellation():
club = config.get_club()
user = User(login="padel.testing@jouf.fr", password="ridicule")
asyncio.run(booking.cancel_booking_id(club, user, 3605033))
def test_cancellation(club, user, booking_filter):
asyncio.run(booking.cancel_booking(club, user, booking_filter))
@patch.dict(
@ -49,9 +38,9 @@ def test_cancellation():
clear=True,
)
def test_main_booking():
court, user = booking.main()
court, player = booking.main()
assert court is not None
assert user.username == "padel.testing@jouf"
assert player.login == "padel.testing@jouf.fr"
@patch.dict(

251
tests/integration_tests/test_connectors.py
View file

@ -3,12 +3,10 @@ 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, Club, User
from models import BookingFilter, Club
from pendulum import DateTime
from yarl import URL
@ -38,9 +36,7 @@ def retrieve_booking_datetime(
{"CLUB_ID": "tpc"},
clear=True,
)
def test_urls():
club = config.get_club()
connector = GestionSportsConnector(club)
def test_urls(connector):
assert (
connector.landing_url
== "https://toulousepadelclub.gestion-sports.com/connexion.php"
@ -65,12 +61,10 @@ def test_urls():
@patch.dict(
os.environ,
{"CLUB_ID": "tpc", "RESOURCES_FOLDER": "/some/path"},
{"RESOURCES_FOLDER": "/some/path"},
clear=True,
)
def test_urls_payload_templates():
club = config.get_club()
connector = GestionSportsConnector(club)
def test_urls_payload_templates(connector):
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"
@ -84,15 +78,8 @@ def test_urls_payload_templates():
)
@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 def test_landing_page(connector):
async with aiohttp.ClientSession() as session:
response = await connector.land(session)
@ -104,17 +91,8 @@ async def test_landing_page():
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 def test_login(connector, user):
async with aiohttp.ClientSession() as session:
await connector.land(session)
@ -130,90 +108,46 @@ async def test_login():
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()
def test_get_booked_court(
connector, booking_success_response, booking_failure_response
):
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),
(601, booking_failure_response),
(602, booking_failure_response),
(603, booking_failure_response),
(614, booking_failure_response),
(605, booking_failure_response),
(606, booking_success_response),
(607, booking_failure_response),
(608, booking_failure_response),
]
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 def test_book_one_court(connector, user, booking_filter):
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
court_id, response = await connector.send_booking_request(
session, pendulum.parse("2024-03-21T13:30:00+01:00"), 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
assert response.get("status") == "ok"
@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")
)
async def test_book(connector, user, booking_filter):
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")
)
def test_build_booking_datetime(connector, booking_filter):
opening_datetime = connector.build_booking_datetime(booking_filter)
assert opening_datetime.year == 2024
assert opening_datetime.month == 3
@ -222,19 +156,8 @@ def test_build_booking_datetime():
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")
)
def test_wait_until_booking_time(mock_now, connector, booking_filter, club):
booking_datetime = retrieve_booking_datetime(booking_filter, club)
seconds = [
@ -252,17 +175,8 @@ def test_wait_until_booking_time(mock_now):
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 def test_get_hash(connector, user):
async with aiohttp.ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
@ -281,17 +195,8 @@ def test_get_hash_input():
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 def test_get_bookings(connector, user):
async with aiohttp.ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
@ -303,17 +208,8 @@ async def test_get_bookings():
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 def test_get_ongoing_bookings(connector, user):
async with aiohttp.ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
@ -322,31 +218,13 @@ async def test_get_ongoing_bookings():
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")
async def test_has_user_ongoing_bookings(connector, user):
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 def test_cancel_booking_id(connector, user):
async with aiohttp.ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
@ -358,73 +236,16 @@ async def test_cancel_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)
def test_find_court(connector):
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)
async def test_cancel_booking(connector, user, booking_filter):
async with aiohttp.ClientSession() as session:
await connector.land(session)
await connector.login(session, user)
await connector.cancel_booking(session, booking_filter)

20
tests/integration_tests/test_gestion_sports_services.py Normal file
View file

@ -0,0 +1,20 @@
import pytest
from gestion_sports_services import GestionSportsServices
@pytest.mark.asyncio
async def test_booking_success(club, user, booking_filter):
court_booked = await GestionSportsServices.book(club, user, booking_filter)
assert court_booked.id is not None
@pytest.mark.asyncio
async def test_user_has_available_slots(club, user):
has_slots = await GestionSportsServices.has_user_available_slots(user, club)
assert has_slots
@pytest.mark.asyncio
async def test_cancel_booking(club, user, booking_filter):
await GestionSportsServices.cancel_booking(user, club, booking_filter)

27
tests/unit_tests/conftest.py
View file

@ -4,6 +4,7 @@ from pathlib import Path
import pendulum
import pytest
from connectors import GestionSportsConnector
from gestion_sports_services import GestionSportsServices
from models import (
BookingFilter,
BookingOpening,
@ -184,6 +185,11 @@ def connector(club) -> GestionSportsConnector:
return GestionSportsConnector(club)
@pytest.fixture
def gs_services() -> GestionSportsServices:
return GestionSportsServices()
@pytest.fixture
def user() -> User:
return User(login="padel.testing@jouf.fr", password="ridicule")
@ -226,6 +232,22 @@ def booking_failure_response() -> dict:
return json.loads(booking_failure_file.read_text(encoding="utf-8"))
@pytest.fixture
def booked_courts_response(
court11,
court12,
court13,
court14,
booking_success_response,
booking_failure_response,
) -> list[tuple[int, dict]]:
court1_resp = court11.id, booking_failure_response
court2_resp = court12.id, booking_failure_response
court3_resp = court13.id, booking_success_response
court4_resp = court14.id, booking_failure_response
return [court1_resp, court2_resp, court3_resp, court4_resp]
@pytest.fixture
def booking_success_from_start(
landing_response,
@ -350,3 +372,8 @@ def cancellation_success_booking_filter() -> BookingFilter:
return BookingFilter(
sport_name="Sport1", date=pendulum.parse("2024-03-21T13:30:00Z")
)
@pytest.fixture
def service() -> GestionSportsServices:
return GestionSportsServices()

83
tests/unit_tests/responses.py Normal file
View file

@ -0,0 +1,83 @@
def make_landing_request_success(aioresponses, connector, landing_response):
aioresponses.get(
connector.landing_url,
status=200,
headers={"Set-Cookie": "PHPSESSID=987512"},
body=landing_response,
)
def make_login_request_fail(aioresponses, connector, login_failure_response):
aioresponses.post(
connector.login_url,
status=200,
payload=login_failure_response,
)
def make_login_request_success(aioresponses, connector, login_success_response):
aioresponses.post(
connector.login_url,
status=200,
headers={"Set-Cookie": "COOK_COMPTE=e2be1;" "COOK_ID_CLUB=22;COOK_ID_USER=666"},
payload=login_success_response,
)
def set_booking_request(aioresponses, connector, booking_response):
aioresponses.post(connector.booking_url, status=200, payload=booking_response)
def set_full_booking_requests_responses(aioresponses, connector, responses_list):
make_landing_request_success(aioresponses, connector, responses_list[0])
make_login_request_success(aioresponses, connector, responses_list[1])
for response in responses_list[2:]:
set_booking_request(aioresponses, connector, response)
def set_ongoing_bookings_response(
aioresponses, connector, user_bookings_get_response, user_bookings_post_response
):
set_hash_response(aioresponses, connector, user_bookings_get_response)
set_bookings_response(aioresponses, connector, user_bookings_post_response)
def set_hash_response(aioresponses, connector, user_bookings_get_response):
aioresponses.get(
connector.user_bookings_url, status=200, body=user_bookings_get_response
)
def set_bookings_response(aioresponses, connector, user_bookings_post_response):
aioresponses.post(
connector.user_bookings_url, status=200, payload=user_bookings_post_response
)
def set_full_user_bookings_responses(aioresponses, connector, responses):
make_landing_request_success(aioresponses, connector, responses[0])
make_login_request_success(aioresponses, connector, responses[1])
set_ongoing_bookings_response(aioresponses, connector, *responses[2:])
def set_cancellation_response(aioresponses, connector, response):
aioresponses.post(connector.booking_cancellation_url, status=200, payload=response)
def set_full_cancellation_by_id_responses(aioresponses, connector, responses):
make_landing_request_success(aioresponses, connector, responses[0])
make_login_request_success(aioresponses, connector, responses[1])
set_hash_response(aioresponses, connector, responses[2])
set_cancellation_response(aioresponses, connector, responses[3])
def set_full_cancellation_responses(aioresponses, connector, responses):
make_landing_request_success(aioresponses, connector, responses[0])
make_login_request_success(aioresponses, connector, responses[1])
# the request to get the hash is made twice
set_hash_response(aioresponses, connector, responses[2])
set_hash_response(aioresponses, connector, responses[2])
set_bookings_response(aioresponses, connector, responses[3])
set_cancellation_response(aioresponses, connector, responses[4])

2
tests/unit_tests/test_config.py
View file

@ -15,7 +15,7 @@ from pendulum import DateTime, Timezone
)
def test_get_booking_filter():
booking_filter = config.get_booking_filter()
assert booking_filter.sport_id == "padel"
assert booking_filter.sport_name == "padel"
assert booking_filter.date == DateTime(
year=2024,
month=2,

205
tests/unit_tests/test_gestion_sports_connector.py
View file

@ -5,90 +5,7 @@ import pytest
from aiohttp import ClientSession
from connectors import GestionSportsConnector
def make_landing_request_success(aioresponses, connector, landing_response):
aioresponses.get(
connector.landing_url,
status=200,
headers={"Set-Cookie": "PHPSESSID=987512"},
body=landing_response,
)
def make_login_request_fail(aioresponses, connector, login_failure_response):
aioresponses.post(
connector.login_url,
status=200,
payload=login_failure_response,
)
def make_login_request_success(aioresponses, connector, login_success_response):
aioresponses.post(
connector.login_url,
status=200,
headers={"Set-Cookie": "COOK_COMPTE=e2be1;" "COOK_ID_CLUB=22;COOK_ID_USER=666"},
payload=login_success_response,
)
def set_booking_request(aioresponses, connector, booking_response):
aioresponses.post(connector.booking_url, status=200, payload=booking_response)
def set_full_booking_requests_responses(aioresponses, connector, responses_list):
make_landing_request_success(aioresponses, connector, responses_list[0])
make_login_request_success(aioresponses, connector, responses_list[1])
for response in responses_list[2:]:
set_booking_request(aioresponses, connector, response)
def set_ongoing_bookings_response(
aioresponses, connector, user_bookings_get_response, user_bookings_post_response
):
set_hash_response(aioresponses, connector, user_bookings_get_response)
set_bookings_response(aioresponses, connector, user_bookings_post_response)
def set_hash_response(aioresponses, connector, user_bookings_get_response):
aioresponses.get(
connector.user_bookings_url, status=200, body=user_bookings_get_response
)
def set_bookings_response(aioresponses, connector, user_bookings_post_response):
aioresponses.post(
connector.user_bookings_url, status=200, payload=user_bookings_post_response
)
def set_full_user_bookings_responses(aioresponses, connector, responses):
make_landing_request_success(aioresponses, connector, responses[0])
make_login_request_success(aioresponses, connector, responses[1])
set_ongoing_bookings_response(aioresponses, connector, *responses[2:])
def set_cancellation_response(aioresponses, connector, response):
aioresponses.post(connector.booking_cancellation_url, status=200, payload=response)
def set_full_cancellation_by_id_responses(aioresponses, connector, responses):
make_landing_request_success(aioresponses, connector, responses[0])
make_login_request_success(aioresponses, connector, responses[1])
set_hash_response(aioresponses, connector, responses[2])
set_cancellation_response(aioresponses, connector, responses[3])
def set_full_cancellation_responses(aioresponses, connector, responses):
make_landing_request_success(aioresponses, connector, responses[0])
make_login_request_success(aioresponses, connector, responses[1])
# the request to get the hash is made twice
set_hash_response(aioresponses, connector, responses[2])
set_hash_response(aioresponses, connector, responses[2])
set_bookings_response(aioresponses, connector, responses[3])
set_cancellation_response(aioresponses, connector, responses[4])
from tests.unit_tests import responses
def test_urls(connector, club):
@ -134,7 +51,7 @@ def test_urls_payload_templates(mock_resources, club):
@pytest.mark.asyncio
async def test_landing_page(aioresponses, connector, landing_response):
make_landing_request_success(aioresponses, connector, landing_response)
responses.make_landing_request_success(aioresponses, connector, landing_response)
async with ClientSession() as session:
response = await connector.land(session)
@ -146,7 +63,9 @@ async def test_landing_page(aioresponses, connector, landing_response):
@pytest.mark.asyncio
async def test_login_success(aioresponses, connector, user, login_success_response):
make_login_request_success(aioresponses, connector, login_success_response)
responses.make_login_request_success(
aioresponses, connector, login_success_response
)
async with ClientSession() as session:
response = await connector.login(session, user)
@ -160,7 +79,7 @@ async def test_login_success(aioresponses, connector, user, login_success_respon
@pytest.mark.asyncio
async def test_login_failure(aioresponses, connector, user, login_failure_response):
make_login_request_fail(aioresponses, connector, login_failure_response)
responses.make_login_request_fail(aioresponses, connector, login_failure_response)
async with ClientSession() as session:
response = await connector.login(session, user)
@ -170,38 +89,9 @@ async def test_login_failure(aioresponses, connector, user, login_failure_respon
assert await response.json() == login_failure_response
@pytest.mark.asyncio
async def test_booking_success(
aioresponses,
connector,
user,
booking_filter,
booking_success_from_start,
):
set_full_booking_requests_responses(
aioresponses, connector, booking_success_from_start
)
court_booked = await connector.book(user, booking_filter)
assert court_booked.id == 2
@pytest.mark.asyncio
async def test_booking_failure(
aioresponses,
connector,
user,
booking_filter,
booking_failure_from_start,
):
set_full_booking_requests_responses(
aioresponses, connector, booking_failure_from_start
)
court_booked = await connector.book(user, booking_filter)
assert court_booked is None
def test_get_booked_court(connector, booked_courts_response):
booked_court = connector.get_booked_court(booked_courts_response, "Sport1")
assert booked_court.number == 3
@pytest.mark.asyncio
@ -212,7 +102,7 @@ async def test_get_ongoing_bookings(
user_bookings_get_response,
user_bookings_list,
):
set_ongoing_bookings_response(
responses.set_ongoing_bookings_response(
aioresponses, connector, user_bookings_get_response, user_bookings_list
)
@ -222,81 +112,19 @@ async def test_get_ongoing_bookings(
assert len(bookings) == 2
@pytest.mark.asyncio
async def test_get_ongoing_bookings(
aioresponses,
connector,
user,
user_bookings_get_response,
user_bookings_list,
):
set_ongoing_bookings_response(
aioresponses, connector, user_bookings_get_response, user_bookings_list
)
async with ClientSession() as session:
bookings = await connector.get_ongoing_bookings(session)
assert len(bookings) == 2
@pytest.mark.asyncio
async def test_has_user_ongoing_bookings(
aioresponses,
connector,
user,
user_has_ongoing_bookings_from_start,
):
set_full_user_bookings_responses(
aioresponses, connector, user_has_ongoing_bookings_from_start
)
has_bookings = await connector.has_user_ongoing_booking(user)
assert has_bookings
@pytest.mark.asyncio
async def test_has_user_ongoing_bookings(
aioresponses,
connector,
user,
user_has_no_ongoing_bookings_from_start,
):
set_full_user_bookings_responses(
aioresponses, connector, user_has_no_ongoing_bookings_from_start
)
has_bookings = await connector.has_user_ongoing_booking(user)
assert not has_bookings
@pytest.mark.asyncio
async def test_cancellation_request(
aioresponses, connector, user_bookings_get_response, cancellation_response
):
set_hash_response(aioresponses, connector, user_bookings_get_response)
set_cancellation_response(aioresponses, connector, cancellation_response)
responses.set_hash_response(aioresponses, connector, user_bookings_get_response)
responses.set_cancellation_response(aioresponses, connector, cancellation_response)
async with ClientSession() as session:
response = await connector.send_cancellation_request(session, 123)
response = await connector.cancel_booking_id(session, 123)
assert await response.json() == cancellation_response
@pytest.mark.asyncio
async def test_cancel_booking_id(
aioresponses, connector, user, cancellation_by_id_from_start
):
set_full_cancellation_by_id_responses(
aioresponses, connector, cancellation_by_id_from_start
)
response = await connector.cancel_booking_id(user, 132)
assert await response.json() == cancellation_by_id_from_start[3]
@pytest.mark.asyncio
async def test_cancel_booking_success(
aioresponses,
@ -305,10 +133,13 @@ async def test_cancel_booking_success(
cancellation_success_booking_filter,
cancellation_success_from_start,
):
set_full_cancellation_responses(
responses.set_full_cancellation_responses(
aioresponses, connector, cancellation_success_from_start
)
response = await connector.cancel_booking(user, cancellation_success_booking_filter)
async with ClientSession() as session:
response = await connector.cancel_booking(
session, cancellation_success_booking_filter
)
assert await response.json() == cancellation_success_from_start[4]

110
tests/unit_tests/test_gestion_sports_services.py Normal file
View file

@ -0,0 +1,110 @@
import pytest
from gestion_sports_services import GestionSportsServices
from tests.unit_tests import responses
@pytest.mark.asyncio
async def test_booking_success(
aioresponses,
connector,
club,
user,
booking_filter,
booking_success_from_start,
):
responses.set_full_booking_requests_responses(
aioresponses, connector, booking_success_from_start
)
court_booked = await GestionSportsServices.book(club, user, booking_filter)
assert court_booked.id == 2
@pytest.mark.asyncio
async def test_booking_failure(
aioresponses,
gs_services,
connector,
club,
user,
booking_filter,
booking_failure_from_start,
):
responses.set_full_booking_requests_responses(
aioresponses, connector, booking_failure_from_start
)
court_booked = await gs_services.book(club, user, booking_filter)
assert court_booked is None
@pytest.mark.asyncio
async def test_user_has_available_booking_slots(
aioresponses,
gs_services,
connector,
user,
club,
user_has_ongoing_bookings_from_start,
):
responses.set_full_user_bookings_responses(
aioresponses, connector, user_has_ongoing_bookings_from_start
)
has_user_available_slots = await gs_services.has_user_available_slots(user, club)
assert has_user_available_slots
@pytest.mark.asyncio
async def test_user_has_no_available_booking_slots(
aioresponses,
gs_services,
connector,
user,
club,
user_has_no_ongoing_bookings_from_start,
):
responses.set_full_user_bookings_responses(
aioresponses, connector, user_has_no_ongoing_bookings_from_start
)
has_user_available_slots = await gs_services.has_user_available_slots(user, club)
assert not has_user_available_slots
@pytest.mark.asyncio
async def test_cancel_booking(
aioresponses,
gs_services,
connector,
user,
club,
booking_filter,
cancellation_success_from_start,
):
responses.set_full_cancellation_responses(
aioresponses, connector, cancellation_success_from_start
)
await gs_services.cancel_booking(user, club, booking_filter)
@pytest.mark.asyncio
async def test_cancel_booking_id(
aioresponses,
gs_services,
connector,
user,
club,
cancellation_success_from_start,
):
responses.set_full_cancellation_responses(
aioresponses, connector, cancellation_success_from_start
)
await gs_services.cancel_booking_id(user, club, 65464)