Merge pull request 'ABle to send the booking request to several courts at the same time' (#5) from feature/book-a-court-among-several into main

Reviewed-on: https://jouf.fr/gitea/stanislas/resa-padel/pulls/5
This commit is contained in:
Stanislas Jouffroy 2024-02-15 21:59:55 +00:00
commit 963ee6b86f
11 changed files with 288 additions and 99 deletions

16
poetry.lock generated
View file

@ -887,6 +887,20 @@ files = [
[package.dependencies]
six = ">=1.5"
[[package]]
name = "python-dotenv"
version = "1.0.1"
description = "Read key-value pairs from a .env file and set them as environment variables"
optional = false
python-versions = ">=3.8"
files = [
{file = "python-dotenv-1.0.1.tar.gz", hash = "sha256:e324ee90a023d808f1959c46bcbc04446a10ced277783dc6ee09987c37ec10ca"},
{file = "python_dotenv-1.0.1-py3-none-any.whl", hash = "sha256:f7b63ef50f1b690dddf550d03497b66d609393b40b564ed0d674909a68ebf16a"},
]
[package.extras]
cli = ["click (>=5.0)"]
[[package]]
name = "pyyaml"
version = "6.0.1"
@ -1137,4 +1151,4 @@ multidict = ">=4.0"
[metadata]
lock-version = "2.0"
python-versions = "^3.10"
content-hash = "38a411ea7014f8fcefa6174669da1a1c807fa45896c69c0a992d3ef52f274b2e"
content-hash = "c85317b4b8334d5d25f97ac426cfa6cd1f63c3c3d0726e313c97cf1de43c985a"

View file

@ -13,6 +13,7 @@ aiohttp = "^3.9.3"
pendulum = "^3.0.0"
pydantic = "^2.6.1"
pydantic-extra-types = "^2.5.0"
python-dotenv = "^1.0.1"
[tool.poetry.group.dev.dependencies]
black = "^24.1.1"
@ -24,6 +25,14 @@ pytest-icdiff = "^0.9"
pytest-asyncio = "^0.23.5"
pytest-aioresponses = "^0.2.0"
[tool.ruff]
line-length = 88
[tool.pytest.ini_options]
pythonpath = [
".", "resa_padel", "rr"
]
[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"

View file

@ -1,33 +1,33 @@
import asyncio
import logging
import pendulum
import config
from aiohttp import ClientSession
from resa_padel import config
from resa_padel.gestion_sports.gestion_sports_connector import GestionSportsConnector
from resa_padel.models import User, BookingFilter
from gestion_sports.gestion_sports_connector import GestionSportsConnector
from models import BookingFilter, User
LOGGER = logging.getLogger(__name__)
async def book(url: str, user: User, booking_filter: BookingFilter) -> ClientSession:
async def book(url: str, user: User, booking_filter: BookingFilter) -> None:
async with ClientSession() as session:
platform = GestionSportsConnector(session, url)
await platform.connect()
await platform.login(user)
await platform.book(booking_filter)
return session
def main() -> None:
LOGGER.info("Starting booking padel court")
LOGGER.info(
f"login={config.USER}, password={config.PASSWORD}, club_id={config.CLUB_ID}"
)
user = User(login=config.USER, password=config.PASSWORD, club_id=config.CLUB_ID)
LOGGER.info(
f"court_id={config.COURT_IDS},sport_id={config.SPORT_ID},date={config.DATE_TIME}"
)
booking_filter = BookingFilter(
court_id=config.COURT_ID,
sport_id=config.SPORT_ID,
date=pendulum.parse(config.DATE_TIME),
court_ids=config.COURT_IDS, sport_id=config.SPORT_ID, date=config.DATE_TIME
)
asyncio.run(book(config.GESTION_SPORTS_URL, user, booking_filter))
LOGGER.info("Finished booking padel court")

View file

@ -1,7 +1,21 @@
import logging.config
import os
import pendulum
import yaml
from dotenv import load_dotenv
load_dotenv()
GESTION_SPORTS_URL = "https://toulousepadelclub.gestion-sports.com"
USER = os.environ.get("USER")
PASSWORD = os.environ.get("PASSWORD")
CLUB_ID = os.environ.get("CLUB_ID")
_court_ids = os.environ.get("COURT_IDS") or ""
COURT_IDS = [int(court_id) for court_id in _court_ids.split(",")] if _court_ids else []
SPORT_ID = int(os.environ.get("SPORT_ID"))
DATE_TIME = pendulum.parse(os.environ.get("DATE_TIME"))
def init_log_config():
@ -11,12 +25,3 @@ def init_log_config():
with open(logging_file, "r") as f:
logging_config = yaml.safe_load(f.read())
logging.config.dictConfig(logging_config)
GESTION_SPORTS_URL = "https://toulousepadelclub.gestion-sports.com"
USER = os.environ.get("USER")
PASSWORD = os.environ.get("PASSWORD")
CLUB_ID = os.environ.get("CLUB_ID")
COURT_ID = os.environ.get("COURT_ID")
SPORT_ID = os.environ.get("SPORT_ID")
DATE_TIME = os.environ.get("DATE_TIME")

View file

@ -1,11 +1,11 @@
import asyncio
import json
import logging
from aiohttp import ClientSession, ClientResponse
from resa_padel.gestion_sports.gestion_sports_payload_builder import (
GestionSportsPayloadBuilder,
)
from resa_padel.models import User
from aiohttp import ClientResponse, ClientSession
from gestion_sports.gestion_sports_payload_builder import \
GestionSportsPayloadBuilder
from models import BookingFilter, User
LOGGER = logging.getLogger(__name__)
HEADERS = {
@ -30,13 +30,21 @@ class GestionSportsConnector:
self.session = session
self.payload_builder = GestionSportsPayloadBuilder()
def __exit__(self, exc_type, exc_val, exc_tb):
self.session.close()
@property
def connection_url(self) -> str:
return f"{self.url}/connexion.php?"
@property
def login_url(self) -> str:
return f"{self.url}/connexion.php?"
@property
def booking_url(self) -> str:
return f"{self.url}/membre/reservation.html?"
async def connect(self) -> ClientResponse:
LOGGER.info("Connecting to GestionSports API")
connection_url = self.url + "/connexion.php?"
async with self.session.get(connection_url) as response:
async with self.session.get(self.connection_url) as response:
await response.text()
return response
@ -48,29 +56,48 @@ class GestionSportsConnector:
.build_login_payload()
)
login_url = f"{self.url}/connexion.php?"
async with self.session.post(
login_url, data=payload, headers=HEADERS
self.login_url, data=payload, headers=HEADERS
) as response:
await response.text()
return response
async def book(self, booking_filter) -> ClientResponse:
async def book(self, booking_filter: BookingFilter) -> int | None:
bookings = await asyncio.gather(
*[
self.book_one_court(booking_filter, court_id)
for court_id in booking_filter.court_ids
],
return_exceptions=True,
)
for court, is_booked in bookings:
if is_booked:
return court
return None
async def book_one_court(
self, booking_filter: BookingFilter, court_id: int
) -> tuple[int, bool]:
date_format = "%d/%m/%Y"
time_format = "%H:%M"
payload = (
self.payload_builder.date(booking_filter.date.date().strftime(date_format))
.time(booking_filter.date.time().strftime(time_format))
.sport_id(booking_filter.sport_id)
.court_id(booking_filter.court_id)
.court_id(court_id)
.build_booking_payload()
)
return court_id, await self.is_court_booked(payload)
booking_url = f"{self.url}/membre/reservation.html?"
async def is_court_booked(self, payload: str) -> bool:
async with self.session.post(
booking_url, data=payload, headers=HEADERS
self.booking_url, data=payload, headers=HEADERS
) as response:
await response.text()
return response
return self.is_response_status_ok(await response.text())
@staticmethod
def is_response_status_ok(response: str) -> bool:
formatted_result = response.removeprefix('"').removesuffix('"')
result_json = json.loads(formatted_result)
return result_json["status"] == "ok"

View file

@ -1,4 +1,4 @@
from resa_padel.exceptions import ArgumentMissing
from exceptions import ArgumentMissing
class GestionSportsPayloadBuilder:
@ -11,31 +11,31 @@ class GestionSportsPayloadBuilder:
self._sport_id = None
self._court_id = None
def login(self, login):
def login(self, login: str):
self._login = login
return self
def password(self, password):
def password(self, password: str):
self._password = password
return self
def club_id(self, club_id):
def club_id(self, club_id: str):
self._club_id = club_id
return self
def date(self, date):
def date(self, date: str):
self._date = date
return self
def time(self, time):
def time(self, time: str):
self._time = time
return self
def sport_id(self, sport_id):
def sport_id(self, sport_id: int):
self._sport_id = sport_id
return self
def court_id(self, court_id):
def court_id(self, court_id: int):
self._court_id = court_id
return self

View file

@ -1,3 +1,5 @@
from typing import List
from pydantic import BaseModel, Field
from pydantic_extra_types.pendulum_dt import DateTime
@ -9,6 +11,6 @@ class User(BaseModel):
class BookingFilter(BaseModel):
court_id: int = Field()
court_ids: List[int] = Field()
sport_id: int = Field()
date: DateTime = Field()

80
tests/fixtures.py Normal file
View file

@ -0,0 +1,80 @@
import json
import pendulum
import pytest
from aiohttp import ClientSession
from resa_padel.gestion_sports.gestion_sports_connector import \
GestionSportsConnector
from resa_padel.gestion_sports.gestion_sports_payload_builder import \
GestionSportsPayloadBuilder
from resa_padel.models import BookingFilter, User
user = User(login="padel.testing@jouf.fr", password="ridicule", club_id="123")
courts = [606, 607, 608]
sport_id = 217
booking_date = pendulum.now().add(days=6).set(hour=18, minute=0, second=0)
booking_filter = BookingFilter(court_ids=courts, sport_id=sport_id, date=booking_date)
booking_failure_response = json.dumps(
{
"status": "error",
"message": "Désolé mais vous avez 1 réservation en cours au Padel en heures pleines et le réglement"
" n'autorise qu'une réservation en heures pleines à la fois au Padel!",
}
)
booking_success_response = json.dumps(
{
"status": "ok",
"message": "Merci, votre réservation s'est bien effectuée, vous allez recevoir un email"
" avec le récapitulatif de votre réservation, pensez à le conserver.",
"id_resa": 3503741,
}
)
date_format = "%d/%m/%Y"
time_format = "%H:%M"
booking_payload = (
GestionSportsPayloadBuilder()
.date(booking_date.date().strftime(date_format))
.time(booking_date.time().strftime(time_format))
.sport_id(sport_id)
.court_id(courts[0])
.build_booking_payload()
)
session = ClientSession()
gestion_sports_url = "https://toulousepadelclub.gestion-sports.com"
gs_connector = GestionSportsConnector(session, gestion_sports_url)
@pytest.fixture
def a_user() -> User:
return user
@pytest.fixture
def a_booking_filter() -> BookingFilter:
return booking_filter
@pytest.fixture
def a_booking_success_response() -> str:
return booking_success_response
@pytest.fixture
def a_booking_failure_response() -> str:
return booking_failure_response
@pytest.fixture
def a_booking_payload() -> str:
return booking_payload
@pytest.fixture
def a_connector():
return gs_connector

View file

@ -1,26 +1,14 @@
import json
import pendulum
import pytest
from aiohttp import ClientSession
from yarl import URL
from resa_padel.gestion_sports.gestion_sports_connector import GestionSportsConnector
from resa_padel.models import User, BookingFilter
from resa_padel.gestion_sports.gestion_sports_connector import \
GestionSportsConnector
from tests.fixtures import (a_booking_failure_response, a_booking_filter,
a_booking_payload, a_booking_success_response,
a_connector, a_user)
gestion_sports_url = "https://toulousepadelclub.gestion-sports.com"
test_user = "padel.testing@jouf.fr"
test_user_id = "232382"
test_password = "ridicule"
test_club_id = "88"
user = User(login=test_user, password=test_password, club_id=test_club_id)
test_court_id = "607"
now = pendulum.now()
date_to_book = now.add(days=6)
datetime_to_book = date_to_book.set(hour=18, minute=0, second=0)
padel_id = 217
court_id = 607
@pytest.mark.asyncio
@ -43,43 +31,86 @@ async def test_should_connect_to_gestion_sports_website():
@pytest.mark.asyncio
async def test_should_login_to_gestion_sports_website():
async def test_should_login_to_gestion_sports_website(a_user):
async with ClientSession() as session:
gs_connector = GestionSportsConnector(session, gestion_sports_url)
await gs_connector.connect()
response = await gs_connector.login(user)
response = await gs_connector.login(a_user)
assert response.status == 200
assert response.request_info.url == URL(gestion_sports_url + "/connexion.php")
assert response.request_info.method == "POST"
cookies = session.cookie_jar.filter_cookies(URL(gestion_sports_url))
assert cookies.get("COOK_ID_CLUB").value == test_club_id
assert cookies.get("COOK_ID_USER").value == test_user_id
assert cookies.get("COOK_ID_CLUB").value is not None
assert cookies.get("COOK_ID_USER").value is not None
assert cookies.get("PHPSESSID") is not None
@pytest.mark.asyncio
async def test_should_book_a_court_from_gestion_sports():
async def test_booking_url_should_be_reachable(a_user, a_booking_filter):
async with ClientSession() as session:
gs_connector = GestionSportsConnector(session, gestion_sports_url)
await gs_connector.connect()
await gs_connector.login(user)
await gs_connector.login(a_user)
booking_filter = BookingFilter(
court_id=court_id, sport_id=padel_id, date=datetime_to_book
)
response = await gs_connector.book(booking_filter)
court_booked = await gs_connector.book(a_booking_filter)
# At 18:00 no chance to get a booking, any day of the week
assert court_booked is None
assert response.status == 200
assert response.request_info.url == URL(
gestion_sports_url + "/membre/reservation.html?"
)
assert response.request_info.method == "POST"
result = await response.text()
formatted_result = result.removeprefix('"').removesuffix('"')
result_json = json.loads(formatted_result)
# booking any day at 18:00 should always be an error if everything goes well ;)
assert result_json["status"] == "error"
@pytest.mark.asyncio
async def test_should_book_a_court_from_gestion_sports(
aioresponses,
a_booking_filter,
a_booking_success_response,
a_booking_failure_response,
):
booking_url = URL(gestion_sports_url + "/membre/reservation.html?")
# first booking request will fail
aioresponses.post(URL(booking_url), status=200, body=a_booking_failure_response)
# first booking request will succeed
aioresponses.post(URL(booking_url), status=200, body=a_booking_success_response)
# first booking request will fail
aioresponses.post(URL(booking_url), status=200, body=a_booking_failure_response)
async with ClientSession() as session:
gs_connector = GestionSportsConnector(session, gestion_sports_url)
court_booked = await gs_connector.book(a_booking_filter)
# the second element of the list is the booked court
assert court_booked == a_booking_filter.court_ids[1]
def test_response_status_should_be_ok(a_booking_success_response):
is_booked = GestionSportsConnector.is_response_status_ok(a_booking_success_response)
assert is_booked
def test_response_status_should_be_not_ok(aioresponses, a_booking_failure_response):
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_connector, a_booking_payload, a_booking_failure_response
):
aioresponses.post(
URL(a_connector.booking_url), status=200, body=a_booking_failure_response
)
is_booked = await a_connector.is_court_booked(a_booking_payload)
assert not is_booked
@pytest.mark.asyncio
async def test_court_should_be_booked(
aioresponses, a_connector, a_booking_payload, a_booking_success_response
):
aioresponses.post(
URL(a_connector.booking_url), status=200, body=a_booking_success_response
)
is_booked = await a_connector.is_court_booked(a_booking_payload)
assert is_booked

View file

@ -1,8 +1,7 @@
import pendulum
from resa_padel.gestion_sports.gestion_sports_payload_builder import (
GestionSportsPayloadBuilder,
)
from resa_padel.gestion_sports.gestion_sports_payload_builder import \
GestionSportsPayloadBuilder
def test_login_payload_should_be_built():

View file

@ -1,10 +1,14 @@
import asyncio
import pendulum
from aioresponses import aioresponses
from resa_padel import booking
from resa_padel.models import BookingFilter, User
from tests.fixtures import (a_booking_failure_response,
a_booking_success_response)
user = "user"
login = "user"
password = "password"
club_id = "98"
court_id = "11"
@ -14,11 +18,14 @@ court_id = "11"
# check that called are passed to the given urls
# check made with cookies, but at the current time, cookies from the response are
# not set in the session. Why ? I don't know....
def test_booking_does_the_rights_calls():
def test_booking_does_the_rights_calls(
a_booking_success_response, a_booking_failure_response
):
# mock connection to the booking platform
booking_url = "https://some.url"
connection_url = booking_url + "/connexion.php"
platform_url = "https://some.url"
connection_url = platform_url + "/connexion.php"
login_url = connection_url
booking_url = platform_url + "/membre/reservation.html"
loop = asyncio.get_event_loop()
@ -26,19 +33,34 @@ def test_booking_does_the_rights_calls():
aio_mock.get(
connection_url,
status=200,
headers={"Set-Cookie": f"connection_called=True; Domain={booking_url}"},
headers={"Set-Cookie": f"connection_called=True; Domain={platform_url}"},
)
aio_mock.post(
login_url,
status=200,
headers={"Set-Cookie": f"login_called=True; Domain={booking_url}"},
headers={"Set-Cookie": f"login_called=True; Domain={platform_url}"},
)
aio_mock.post(
booking_url,
status=200,
headers={"Set-Cookie": f"booking_called=True; Domain={platform_url}"},
body=a_booking_failure_response,
)
aio_mock.post(
booking_url,
status=200,
headers={"Set-Cookie": f"booking_called=True; Domain={platform_url}"},
body=a_booking_success_response,
)
session = loop.run_until_complete(
booking.book(booking_url, user, password, club_id, court_id)
user = User(login=login, password=password, club_id=club_id)
booking_filter = BookingFilter(
court_ids=[607, 606], sport_id=444, date=pendulum.now().add(days=6)
)
cookies = session.cookie_jar.filter_cookies(booking_url)
loop.run_until_complete(booking.book(platform_url, user, booking_filter))
# cookies = session.cookie_jar.filter_cookies(platform_url)
# assert cookies.get("connection_called") == "True"
# assert cookies.get("login_called") == "True"
# assert cookies.get("booking_called") == "True"